From ad080c4c14e6841b9863398746f93339ac60752b Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Thu, 30 Sep 2021 17:41:23 -0700 Subject: [PATCH 1/2] Filter out duplicate pending acks Seeing reports of long processing delays in the NSE. We can fetch messages faster than they can be acked. This backs up our ack queue with a bunch of repeat work, which means we re-fetch the same envelopes over and over, which fills the ack queue with more repeat work. --- .../src/Messages/MessageFetcherJob.swift | 35 ++++++++++++++----- SignalServiceKit/src/Util/OWSOperation.h | 3 ++ SignalServiceKit/src/Util/OWSOperation.m | 8 +++++ 3 files changed, 38 insertions(+), 8 deletions(-) diff --git a/SignalServiceKit/src/Messages/MessageFetcherJob.swift b/SignalServiceKit/src/Messages/MessageFetcherJob.swift index 6b9d8aba81..49c7c7caa7 100644 --- a/SignalServiceKit/src/Messages/MessageFetcherJob.swift +++ b/SignalServiceKit/src/Messages/MessageFetcherJob.swift @@ -562,6 +562,21 @@ private class MessageAckOperation: OWSOperation { private let envelopeInfo: EnvelopeInfo private let pendingAck: PendingTask + // A heuristic to quickly filter out multiple ack attempts for the same message + // This doesn't affect correctness, just tries to guard against backing up our operation queue with repeat work + static private var inFlightAcks = AtomicSet() + private var didRecordAckId = false + private var inFlightAckId: String { + if let serverGuid = envelopeInfo.serverGuid?.nilIfEmpty { + return serverGuid + } else if let sourceUuid = envelopeInfo.sourceAddress?.uuid { + return "\(sourceUuid.uuidString)_\(envelopeInfo.timestamp)" + } else { + // This *could* collide, but we don't have enough info to ack the message anyway. So it should be fine. + return "\(envelopeInfo.serviceTimestamp)" + } + } + fileprivate required init(envelopeInfo: EnvelopeInfo, pendingAck: PendingTask) { self.envelopeInfo = envelopeInfo @@ -574,6 +589,13 @@ private class MessageAckOperation: OWSOperation { // MessageAckOperation must have a higher priority than than the // operations used to flush the ack operation queue. self.queuePriority = .high + if Self.inFlightAcks.contains(inFlightAckId) { + Logger.info("Cancelling new ack operation for \(envelopeInfo). Duplicate ack already enqueued") + self.cancel() + } else { + Self.inFlightAcks.insert(inFlightAckId) + didRecordAckId = true + } } public override func run() { @@ -611,14 +633,11 @@ private class MessageAckOperation: OWSOperation { } @objc - public override func didSucceed() { - super.didSucceed() - pendingAck.complete() - } - - @objc - public override func didFail(error: Error) { - super.didFail(error: error) + public override func didComplete() { + super.didComplete() + if didRecordAckId { + Self.inFlightAcks.remove(inFlightAckId) + } pendingAck.complete() } } diff --git a/SignalServiceKit/src/Util/OWSOperation.h b/SignalServiceKit/src/Util/OWSOperation.h index df0e95e2c2..c0a6356237 100644 --- a/SignalServiceKit/src/Util/OWSOperation.h +++ b/SignalServiceKit/src/Util/OWSOperation.h @@ -53,6 +53,9 @@ typedef NS_ENUM(NSInteger, OWSOperationState) { // Called at most one time, once retry is no longer possible. - (void)didFailWithError:(NSError *)error NS_SWIFT_NAME(didFail(error:)); +// Called exactly once after operation has moved to OWSOperationStateFinished +- (void)didComplete; + // How long to wait before retry, if possible - (NSTimeInterval)retryInterval; diff --git a/SignalServiceKit/src/Util/OWSOperation.m b/SignalServiceKit/src/Util/OWSOperation.m index 48645fc253..d1a3d92723 100644 --- a/SignalServiceKit/src/Util/OWSOperation.m +++ b/SignalServiceKit/src/Util/OWSOperation.m @@ -111,6 +111,13 @@ NSString *const OWSOperationKeyIsFinished = @"isFinished"; // Override in subclass if necessary } +// Called exactly once after the operation is marked complete, either with success, failure, or cancellation +- (void)didComplete +{ + // no-op + // Override in subclass if necessary +} + #pragma mark - NSOperation overrides - (NSString *)eventId @@ -271,6 +278,7 @@ NSString *const OWSOperationKeyIsFinished = @"isFinished"; [self didChangeValueForKey:OWSOperationKeyIsExecuting]; [self didChangeValueForKey:OWSOperationKeyIsFinished]; + [self didComplete]; [BenchManager completeEventWithEventId:self.eventId]; } From c2cd2cd4c67f05c3f03a4e03c82e506f3f5f5b45 Mon Sep 17 00:00:00 2001 From: Michelle Linington Date: Thu, 30 Sep 2021 19:32:10 -0700 Subject: [PATCH 2/2] PR Feedback --- SignalServiceKit/src/Messages/MessageFetcherJob.swift | 4 ++++ SignalServiceKit/src/Messages/MessageProcessor.swift | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/SignalServiceKit/src/Messages/MessageFetcherJob.swift b/SignalServiceKit/src/Messages/MessageFetcherJob.swift index 49c7c7caa7..9b36590011 100644 --- a/SignalServiceKit/src/Messages/MessageFetcherJob.swift +++ b/SignalServiceKit/src/Messages/MessageFetcherJob.swift @@ -567,6 +567,9 @@ private class MessageAckOperation: OWSOperation { static private var inFlightAcks = AtomicSet() private var didRecordAckId = false private var inFlightAckId: String { + // All messages *should* have a guid, but we'll handle things correctly if they don't + owsAssertDebug(envelopeInfo.serverGuid?.nilIfEmpty != nil) + if let serverGuid = envelopeInfo.serverGuid?.nilIfEmpty { return serverGuid } else if let sourceUuid = envelopeInfo.sourceAddress?.uuid { @@ -591,6 +594,7 @@ private class MessageAckOperation: OWSOperation { self.queuePriority = .high if Self.inFlightAcks.contains(inFlightAckId) { Logger.info("Cancelling new ack operation for \(envelopeInfo). Duplicate ack already enqueued") + pendingAck.complete() self.cancel() } else { Self.inFlightAcks.insert(inFlightAckId) diff --git a/SignalServiceKit/src/Messages/MessageProcessor.swift b/SignalServiceKit/src/Messages/MessageProcessor.swift index e221e61372..43811c28e7 100644 --- a/SignalServiceKit/src/Messages/MessageProcessor.swift +++ b/SignalServiceKit/src/Messages/MessageProcessor.swift @@ -233,7 +233,7 @@ public class MessageProcessor: NSObject { // This is tricky since there are multiple variables (e.g. network // perf affects fetch, CPU perf affects processing). public var hasSomeQueuedContent: Bool { - queuedContentCount >= 25 + queuedContentCount >= 10 } public var queuedContentCount: Int {