* Where the line falls is the subject of O12 in the event-encoding research, and it falls between the events * an application asked for by name and the ones it did not. At {@code IMMEDIATE}, {@code track} and - * {@code identify} write themselves before returning, because an application reporting an error is saying - * this matters more than the microseconds it costs, and the crash it describes may be moments away. They - * encode and write only themselves: what is held, and the summaries, which can take megabytes to encode, are - * left to a commit queued on the commit executor in the same call, so the exposures leading up to the error - * follow it moments later rather than costing the caller's thread. Everything else is committed once + * {@code identify} are commit points, because an application reporting an error is saying this matters more + * than the microseconds it costs, and the crash it describes may be moments away. A commit encodes the whole + * held run, so the exposures leading up to the error go down with it. Everything else is committed once * {@link #PENDING_COMMIT_THRESHOLD} events have accumulated where persistence is on, on {@code flush}, at a * delivery, on backgrounding, or when the application flushes from its own crash handler. Backgrounding * commits through {@code flush}: {@link ConnectivityManager} calls it on every move to the background, and @@ -46,10 +44,8 @@ * {@code DEFERRED} it costs what an evaluation does. *
* Threads. Recording runs on whichever thread evaluated or tracked. Commits run on {@link #commitExecutor}, - * or on the caller's thread for a flush, one at a time under {@link #commitLock}. A delivery commits too, on - * the event processor's delivery thread, before it closes a batch. A {@code track} or {@code identify} at - * {@code IMMEDIATE} writes its one event without {@link #commitLock}, so that it never waits for a commit - * encoding a run. + * or on the caller's thread at a commit point, one at a time under {@link #commitLock}. A delivery commits + * too, on the event processor's delivery thread, before it closes a batch. */ final class EventCommitter { /** @@ -63,15 +59,15 @@ final class EventCommitter { * could take. An uncaught exception runs the application's handler while the process is still alive, and * a handler that flushes commits the whole held run. So this is the window for the terminations that run * nothing on the way out -- {@code SIGKILL}, a native crash, an ANR kill, and the system reclaiming a - * backgrounded process. A {@code track} at {@code IMMEDIATE} narrows it too, because it queues a commit. + * backgrounded process. A {@code track} at {@code IMMEDIATE} closes it too, because it commits. *
- * The second is what a {@code flush} usually costs, since it encodes whatever is held before returning. + * The second is what such a {@code track} usually costs, since it encodes whatever is held before returning. * The commit executor keeps the run drained, so the common case is far below this; raising it trades that * cost against the number of writes. *
* Neither is a ceiling. Reaching this only schedules a commit, and recording does not wait for it, so * where events arrive faster than the commit executor encodes them, or the executor is starved, the run - * grows up to the event capacity, and the next flush encodes all of it. + * grows up to the event capacity, and the next commit point encodes all of it. */ static final int PENDING_COMMIT_THRESHOLD = 32; @@ -90,8 +86,7 @@ final class EventCommitter { */ private final int capacity; /** - * Whether {@code track} and {@code identify} write before returning, and a flush writes even where - * nothing persists, rather than leaving that work to a commit off the caller's thread. + * Whether a commit point encodes and writes before returning, rather than queuing that work. */ private final boolean commitOnCallerThread; /** The event processor's: set once it closes, after which nothing more is recorded. */ @@ -107,10 +102,10 @@ final class EventCommitter { /** * Held for the whole of a commit, so that only one runs at a time. *
- * This is what a flush's guarantee rests on. Without it a commit already in flight could take the - * caller's events out of {@link #pending} before the caller got there, leaving the caller nothing to + * This is what {@code track}'s guarantee rests on. Without it a commit already in flight could take the + * caller's event out of {@link #pending} before the caller got there, leaving the caller nothing to * write and returning while those bytes were still being produced somewhere else. Waiting here instead - * means that when the call returns the events are on disk, whichever commit put them there. + * means that when the call returns the event is on disk, whichever commit put it there. *
* Distinct from {@link #recordLock}, which is only held long enough to hand the run and the counters * over: encoding under this lock must not block a thread that is merely recording. @@ -121,7 +116,7 @@ final class EventCommitter { * The last recording a finished commit covered. Written under {@link #commitLock}, and read outside it * so that a caller whose recording is already durable does not queue for the lock at all. *
- * This is what makes concurrent flushes a group commit. The first caller's
+ * This is what makes concurrent {@code IMMEDIATE} commit points a group commit. The first caller's
* commit takes every recording made before it took the run, including those of callers now waiting
* behind it. They find their recordings covered and return, and the next caller still uncovered
* commits everything that arrived in the meantime. Without it each waiter committed again in turn,
@@ -156,18 +151,6 @@ final class EventCommitter {
*/
private final List
- * Taken there rather than left, so the summarizer's contexts are cleared at each of them as they were when
- * they committed the counters themselves: an application identifying its way through more contexts than
- * capacity faster than the commit executor keeps up would otherwise find the cardinality limit binding.
- * Every commit takes them along with {@link #pending}, so an evaluation's counter and its full event are
- * still staged together.
- */
- private final List
* At {@code DEFERRED} it writes nothing itself (durability ยง13.3), and is committed with what is held
* at the pending threshold, a flush, a delivery, backgrounding or close. A commit per call would cost
@@ -291,59 +275,14 @@ void recordEvaluation(
* {@link #recordLock} from the threads evaluating flags and the store's locks from delivery.
*/
void recordAtCommitPoint(Event event) {
- if (commitOnCallerThread && store.isPersisting()) {
- writeAlone(event);
- } else {
- record(event);
- }
- }
-
- /**
- * Encodes and writes one event on the caller's thread, and queues a commit of everything else.
- *
- * Only the event itself, so the caller pays for one small encode and one write however much is held:
- * a commit encodes the held run and every summary, and a summary with a large context or many flags
- * takes megabytes, which is a frozen frame on the main thread. They follow on the commit executor,
- * queued here. The event is written ahead of the run held before it, which LaunchDarkly does not mind:
- * every event carries its own creation date.
- *
- * Not under {@link #commitLock}, which a commit holds for the whole of its encoding, so this never waits
- * for one. The store writes the one frame rather than committing, which would also write whatever a
- * commit in flight has staged so far, and the encoder shares nothing between threads; the one wait left
- * is for the store's write lock, held by a write or by a delivery reading a batch back.
- */
- private void writeAlone(Event event) {
- try {
- boolean admitted;
- synchronized (recordLock) {
- // As in record: orders this against close()'s final commit, which this does not join, so an
- // event admitted here is written before the store is released, and one refused never is.
- if (closed.get()) {
- return;
- }
- admitted = admit(event);
- if (admitted) {
- // Reserved where it is admitted, as a run is where it leaves pending, so capacity counts
- // it while it is encoded and a recording on another thread cannot take its place.
- store.reserve(1);
- }
- takenSummaries.addAll(eventBuffer.takeSummaries());
- summaryContextsExceeded.set(false);
- }
- if (admitted) {
- byte[] serialized = eventBuffer.serialize(event);
- if (serialized == null) {
- // Counted by the commit queued below, which takes the failure.
- store.releaseReservations(1);
- } else if (!store.writeReservedAlone(serialized)) {
- // Only the bytes are checked: a frame or byte capacity it turns out not to fit.
- reportRefusedByStore("event", serialized);
- }
+ long sequence = record(event);
+ if (commitOnCallerThread) {
+ try {
+ commitDurablyThrough(sequence);
+ } catch (RuntimeException e) {
+ // As in recordEvaluation: on the caller's thread, so a failure is logged, not thrown.
+ logUnexpectedError(e);
}
- scheduleCommit();
- } catch (RuntimeException e) {
- // As in recordEvaluation: on the caller's thread, so a failure is logged, not thrown.
- logUnexpectedError(e);
}
}
@@ -534,12 +473,6 @@ void commitDurably() {
run = pending.isEmpty() ? Collections.emptyList() : new ArrayList<>(pending);
pending.clear();
summaries = eventBuffer.takeSummaries();
- if (!takenSummaries.isEmpty()) {
- List
- * Each {@link #stageReserved} or {@link #writeReservedAlone} uses one up, and
- * {@link #releaseReservations} gives back any that never made it, such as an event that could not be
- * serialized.
+ * Each {@link #stageReserved} uses one up, and {@link #releaseReservations} gives back any that
+ * never made it, such as an event that could not be serialized.
*/
void reserve(int events) {
synchronized (bufferLock) {
@@ -338,14 +337,10 @@ void reserve(int events) {
}
}
- /**
- * Gives back {@code events} reservations not used up by {@link #stageReserved} or
- * {@link #writeReservedAlone}. Only the caller's own, since another caller may hold reservations
- * alongside.
- */
- void releaseReservations(int events) {
+ /** Gives back every reservation not used up by {@link #stageReserved}. */
+ void releaseReservations() {
synchronized (bufferLock) {
- reservedEvents = Math.max(0, reservedEvents - events);
+ reservedEvents = 0;
}
}
@@ -390,9 +385,6 @@ boolean stageReserved(byte[] serializedEvent) {
private boolean stage(byte[] serializedEvent, boolean bypassingCapacity, boolean reserved) {
if (serializedEvent == null || serializedEvent.length == 0
|| serializedEvent.length > EventLogFormat.MAX_FRAME_SIZE) {
- if (reserved) {
- releaseReservations(1);
- }
return false;
}
long frameSize = (long) serializedEvent.length + EventLogFormat.FRAME_OVERHEAD;
@@ -525,56 +517,6 @@ void commit() {
}
}
- /**
- * Writes one serialized event to the log on its own, using up one reservation as
- * {@link #stageReserved} does, but leaving what anything else has staged to the commit that staged it,
- * so the caller pays for writing its own bytes and no one else's. It may land ahead of events staged
- * before it.
- *
- * Without persistence, or once it has failed, the event is staged in memory instead, as a commit
- * leaves it.
- *
- * @return false if the event is too large for a frame or for the bytes left in the store
- */
- boolean writeReservedAlone(byte[] serializedEvent) {
- // As in commit: not worth waiting on ioLock for where there is nothing to write.
- if (!persistEvents) {
- return stageReserved(serializedEvent);
- }
- if (serializedEvent == null || serializedEvent.length == 0
- || serializedEvent.length > EventLogFormat.MAX_FRAME_SIZE) {
- releaseReservations(1);
- return false;
- }
- ByteArrayOutputStream framed =
- new ByteArrayOutputStream(serializedEvent.length + EventLogFormat.FRAME_OVERHEAD);
- EventLogFormat.writeFrame(framed, serializedEvent);
- byte[] frame = framed.toByteArray();
- synchronized (ioLock) {
- Staged alone;
- synchronized (bufferLock) {
- // In the same critical section as the count that replaces it, so the event is never
- // counted twice or not at all.
- if (reservedEvents > 0) {
- reservedEvents--;
- }
- if (countedBytesHoldingBufferLock() + frame.length > eventByteCapacity) {
- return false;
- }
- if (!persistEvents) {
- writeQuietly(bufferData, frame);
- bufferedEventCount++;
- return true;
- }
- alone = new Staged(frame, frame.length, 1, 0);
- inTransitEvents++;
- inTransitBytes += frame.length;
- }
- writeHoldingIoLock(alone);
- }
- return true;
- }
-
// ---- delivery ----
/**
@@ -1133,11 +1075,6 @@ private void commitHoldingIoLock() {
}
staged = takeStagedHoldingBufferLock();
}
- writeHoldingIoLock(staged);
- }
-
- /** Requires {@code ioLock}, and {@code staged} counted in transit. */
- private void writeHoldingIoLock(Staged staged) {
byte[] bytes = staged.frames;
int written = 0;
diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventPersistence.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventPersistence.java
index 785c7a40..387a53a5 100644
--- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventPersistence.java
+++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/integrations/EventPersistence.java
@@ -43,11 +43,13 @@ public enum EventPersistence {
/**
* Events are written to disk before {@code track} and {@code identify} return.
*
- * The write happens on the calling thread and covers that one event, costing tens to hundreds of
- * microseconds depending on the device, however many evaluations are waiting. Those, and the summaries
- * of them, are written on a thread of the SDK's that the call sets going, so they follow moments later.
- * In exchange there is no window: a process that dies the instant after {@code track} returns still
- * reports that event on the next run. Evaluating a flag stays in memory regardless.
+ * The write happens on the calling thread, costing tens to hundreds of microseconds depending on the
+ * device and on how many evaluations are waiting to be encoded alongside it. Evaluations are usually
+ * written in the background before many build up. A burst of them faster than that can leave up to the
+ * event capacity waiting, and the {@code track} or {@code identify} that follows encodes them all, which
+ * can take milliseconds. In exchange there is no window: a process that dies the instant after
+ * {@code track} returns still reports that event on the next run. Evaluating a flag stays in memory
+ * regardless.
*
* Because this is disk I/O on the calling thread, an application that calls {@code track} or
* {@code identify} on the main thread with {@link android.os.StrictMode} detecting disk writes will see
diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java
index 0489e132..29dabbea 100644
--- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java
+++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java
@@ -228,9 +228,9 @@ public void immediatePersistencePutsATrackOnDiskBeforeItReturns() throws Excepti
}
@Test
- public void anImmediateTrackWritesOnlyItselfAndLeavesTheRestToTheCommitExecutor() throws Exception {
- // A summary can take megabytes to encode, which on the main thread is a frozen frame, so a track writes
- // its own event and queues the rest. Holding the commit executor shows which write is whose.
+ public void anImmediateTrackPutsTheEvaluationsBeforeItOnDiskBeforeItReturns() throws Exception {
+ // The exposures leading up to an error are what a kill right after the track would otherwise take.
+ // Holding the commit executor leaves the track's own write as the only one there can be.
ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor();
ExecutorService diagnosticExecutor = EventUtil.makeDiagnosticsTaskExecutor();
diagnosticExecutors.add(diagnosticExecutor);
@@ -248,12 +248,6 @@ public void anImmediateTrackWritesOnlyItselfAndLeavesTheRestToTheCommitExecutor(
eventProcessor.recordCustomEvent(CONTEXT, "fatal-error", LDValue.of("boom"), null);
assertEquals(1, kindsOnDisk("custom"));
- assertEquals(0, kindsOnDisk("feature"));
- assertEquals(0, kindsOnDisk("summary"));
-
- commitsResume.countDown();
- lastCommitExecutor.submit(() -> { }).get(5, TimeUnit.SECONDS);
-
assertEquals(1, kindsOnDisk("feature"));
assertEquals(1, kindsOnDisk("summary"));
} finally {
@@ -262,50 +256,6 @@ public void anImmediateTrackWritesOnlyItselfAndLeavesTheRestToTheCommitExecutor(
}
}
- @Test
- public void anImmediateTrackDoesNotWaitForACommitInProgress() throws Exception {
- // The commit executor's commit is stalled while it holds the commit lock, as one encoding a large
- // summary would be. The track has nothing to wait for it about.
- CountDownLatch commitStalled = new CountDownLatch(1);
- CountDownLatch commitResumes = new CountDownLatch(1);
- EventStore store = new EventStore(EventDirectory.defaultDirectory(eventsDirectory.getRoot(), MOBILE_KEY),
- DEFAULT_CAPACITY, true, logging.logger) {
- @Override
- void commit() {
- if (Thread.currentThread().getName().startsWith("LaunchDarkly-event-store")) {
- commitStalled.countDown();
- awaitQuietly(commitResumes, 10, TimeUnit.SECONDS);
- }
- super.commit();
- }
- };
- ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor();
- ExecutorService diagnosticExecutor = EventUtil.makeDiagnosticsTaskExecutor();
- diagnosticExecutors.add(diagnosticExecutor);
- DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(),
- UNUSED_EVENTS_URI, null, NO_PERIODIC_FLUSH_MILLIS, 60_000, CLOSE_BUDGET_MILLIS,
- scheduler, diagnosticExecutor, store, DEFAULT_CAPACITY, true);
- try {
- for (int i = 0; i < EventCommitter.PENDING_COMMIT_THRESHOLD; i++) {
- eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, VARIATION,
- FLAG_VALUE, EvaluationReason.off(), DEFAULT_VALUE, true, null);
- }
- assertTrue("the threshold's commit never started", commitStalled.await(5, TimeUnit.SECONDS));
-
- long start = System.nanoTime();
- eventProcessor.recordCustomEvent(CONTEXT, "fatal-error", LDValue.of("boom"), null);
- long tookMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
-
- assertTrue("the track waited " + tookMillis + "ms for the commit in progress", tookMillis < 1_000);
- // Without what the stalled commit had staged before it stalled, which is that commit's to write.
- assertEquals(1, kindsOnDisk("custom"));
- assertEquals(0, kindsOnDisk("feature"));
- } finally {
- commitResumes.countDown();
- eventProcessor.close();
- }
- }
-
@Test
public void aDeferredTrackWritesNothingItselfAndReachesTheDiskAtThePendingThreshold() throws Exception {
EventStore store = EventStore.create(eventsDirectory.getRoot(), MOBILE_KEY,
@@ -1472,20 +1422,28 @@ public File getNoBackupFilesDir() {
}
@Test
- public void concurrentImmediateTracksAreEachDurableOnReturn() throws Exception {
+ public void concurrentImmediateTracksShareCommitsAndEachIsDurableOnReturn() throws Exception {
final int trackers = 8;
+ Queue