Skip to content

fix: Commit everything held at a track or identify at immediate persistence #419

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abelonogov-ld wants to merge 1 commit into
base: andrey/event-durability-tier3-persistence
Choose a base branch
from
+66 −296
Open

fix: Commit everything held at a track or identify at immediate persistence #419

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 34 additions & 114 deletions ...hdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventCommitter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,11 +29,9 @@
* <p>
* 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
Expand All @@ -46,10 +44,8 @@
* {@code DEFERRED} it costs what an evaluation does.
* <p>
* 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 {
/**
Expand All @@ -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.
* <p>
* 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.
* <p>
* 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;

Expand All @@ -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. */
Expand All @@ -107,10 +102,10 @@ final class EventCommitter {
/**
* Held for the whole of a commit, so that only one runs at a time.
* <p>
* 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.
* <p>
* 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.
Expand All @@ -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.
* <p>
* 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,
Expand Down Expand Up @@ -156,18 +151,6 @@ final class EventCommitter {
*/
private final List<Event> pending = new ArrayList<>();

/**
* Counters a {@code track} or {@code identify} at {@code IMMEDIATE} took from the summarizer for the next
* commit to encode, guarded by {@link #recordLock}.
* <p>
* 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<EventSummarizer.EventSummary> takenSummaries = new ArrayList<>();

/**
* Numbers every recording, guarded by {@link #recordLock}, so a caller at a commit point can tell whether
* a commit has already covered it.
Expand Down Expand Up @@ -270,7 +253,8 @@ void recordEvaluation(
}
// Deliberately no commit point. An evaluation is expected to cost what a map lookup costs, and it
// is usually the main thread doing it. The store writes these on its own thread once enough of them
// have piled up, and at IMMEDIATE the next track or identify queues that write as well.
// have piled up, and at IMMEDIATE the next event recorded at a commit point makes them durable along
// with itself.
if (needsCommit) {
scheduleCommitWherePersisting();
}
Expand All @@ -283,67 +267,22 @@ void recordEvaluation(
}

/**
* Records an {@code identify} or {@code track} event, and at {@code IMMEDIATE} writes it before returning.
* Records an {@code identify} or {@code track} event, and at {@code IMMEDIATE} commits before returning.
* <p>
* 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
* an application that tracks every few seconds one encode and write per event, each taking
* {@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.
* <p>
* 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.
* <p>
* 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);
}
}

Expand Down Expand Up @@ -534,20 +473,18 @@ void commitDurably() {
run = pending.isEmpty() ? Collections.emptyList() : new ArrayList<>(pending);
pending.clear();
summaries = eventBuffer.takeSummaries();
if (!takenSummaries.isEmpty()) {
List<EventSummarizer.EventSummary> taken = new ArrayList<>(takenSummaries);
taken.addAll(summaries);
takenSummaries.clear();
summaries = taken;
}
summaryContextsExceeded.set(false);
// Reserved in the same critical section the run leaves pending in, so capacity counts
// it in one place or the other throughout the encoding. Each staged event uses up its
// reservation as it arrives, so none is counted twice either.
store.reserve(run.size());
}
try {
stageRun(run);
try {
stageRun(run);
} finally {
store.releaseReservations();
}
stageSummaries(summaries);
} catch (Throwable t) {
// The run and the counters have left pending, and nothing puts them back, so a flush must not
Expand Down Expand Up @@ -681,17 +618,10 @@ void shutdown() {
* they finished encoding, which is not the order they were recorded in.
*/
private void stageRun(List<Event> run) {
int unused = run.size();
try {
for (byte[] serialized : eventBuffer.serializeAll(run)) {
unused--;
if (!store.stageReserved(serialized)) {
reportRefusedByStore("event", serialized);
}
for (byte[] serialized : eventBuffer.serializeAll(run)) {
if (!store.stageReserved(serialized)) {
reportRefusedByStore("event", serialized);
}
} finally {
// Only the run's own: a track or identify writing itself alone may hold one beside them.
store.releaseReservations(unused);
}
}

Expand Down Expand Up @@ -730,21 +660,11 @@ private void reportRefusedByStore(String what, byte[] serialized) {
* {@link #recordLock}.
*/
private void addPending(Event event) {
if (admit(event)) {
pending.add(event);
}
}

/**
* Whether to keep an event, counting it as dropped if there is no room for it. Requires
* {@link #recordLock}.
*/
private boolean admit(Event event) {
// Ahead of the capacity check, because an event the SDK was never going to send is not a
// loss and must not be counted as one. Sampling and capacity are different reasons not to
// keep an event, and only the second is one the SDK owes anyone a count of.
if (!Sampler.shouldSample(event.getSamplingRatio())) {
return false;
return;
}
// The bytes are checked here too, before the event is encoded, so that one there is no room for
// costs nothing; whether it fits is decided when it is staged.
Expand All @@ -753,13 +673,13 @@ private boolean admit(Event event) {
logger.warn("Exceeded event queue capacity. Increase capacity to avoid dropping events.");
}
droppedEvents.incrementAndGet();
return false;
return;
}
// Read before written: a write on every event would contend for the line with nothing to change.
if (capacityExceeded.get()) {
capacityExceeded.set(false);
}
return true;
pending.add(event);
}

/**
Expand Down
73 changes: 5 additions & 68 deletions launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -328,24 +328,19 @@ boolean admitsEvent(int heldElsewhere, int capacity) {
* Counts events against capacity before they are staged, for a caller that has taken them from
* somewhere capacity was already counting them and will encode them before staging.
* <p>
* 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) {
reservedEvents += 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;
}
}

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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.
* <p>
* 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 ----

/**
Expand Down Expand Up @@ -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;
Expand Down
Loading
Loading