From 8bb8c5344367e441581eba76d2e18c3c5c076729 Mon Sep 17 00:00:00 2001 From: Gaston Thea Date: Wed, 24 Sep 2025 10:29:37 -0300 Subject: [PATCH 1/2] Remove blockingqueue in spliteventsmanager --- Split/Events/SplitEventsManager.swift | 46 ++++++--------------------- 1 file changed, 9 insertions(+), 37 deletions(-) diff --git a/Split/Events/SplitEventsManager.swift b/Split/Events/SplitEventsManager.swift index d78c70f11..1ad2a1d78 100644 --- a/Split/Events/SplitEventsManager.swift +++ b/Split/Events/SplitEventsManager.swift @@ -27,16 +27,14 @@ class DefaultSplitEventsManager: SplitEventsManager { private let processQueue: DispatchQueue private let dataAccessQueue: DispatchQueue private var isStarted: Bool - private var eventsQueue: InternalEventBlockingQueue init(config: SplitClientConfig) { - self.processQueue = DispatchQueue(label: "split-evt-mngr-process", attributes: .concurrent) + self.processQueue = DispatchQueue(label: "split-evt-mngr-process") self.dataAccessQueue = DispatchQueue(label: "split-evt-mngr-data", target: DispatchQueue.general) self.isStarted = false self.sdkReadyTimeStart = Date().unixTimestampInMiliseconds() self.readingRefreshTime = 300 self.triggered = [SplitInternalEvent]() - self.eventsQueue = DefaultInternalEventBlockingQueue() self.executionTimes = [String: Int]() registerMaxAllowedExecutionTimesPerEvent() @@ -51,10 +49,9 @@ class DefaultSplitEventsManager: SplitEventsManager { func notifyInternalEvent(_ event: SplitInternalEvent) { processQueue.async { [weak self] in - if let self = self { - Logger.v("Event \(event) notified") - self.eventsQueue.add(event) - } + guard let self = self else { return } + Logger.v("Event \(event) notified") + self.processEvent(event) } } @@ -78,11 +75,6 @@ class DefaultSplitEventsManager: SplitEventsManager { } self.isStarted = true } - processQueue.async { [weak self] in - if let self = self { - self.processEvents() - } - } } func eventAlreadyTriggered(event: SplitEvent) -> Bool { @@ -98,10 +90,6 @@ class DefaultSplitEventsManager: SplitEventsManager { guard let self = self else { return } self.isStarted = false self.subscriptions.removeAll() - self.processQueue.sync { - self.eventsQueue.stop() - self.eventsQueue.stop() - } } } @@ -127,30 +115,15 @@ class DefaultSplitEventsManager: SplitEventsManager { return isRunning } - private func takeEvent() -> SplitInternalEvent? { - do { - return try eventsQueue.take() - } catch BlockingQueueError.hasBeenStopped { - Logger.d("Events manager stoped") - } catch BlockingQueueError.noElementAvailable { - return nil - } catch { - Logger.d("Events manager take event has exit because \(error.localizedDescription)") - } - return nil - } + private func processEvent(_ event: SplitInternalEvent) { + guard isRunning() else { return } - private func processEvents() { - while isRunning() { - guard let event = takeEvent() else { - continue - } - self.triggered.append(event) + self.triggered.append(event) switch event { case .splitsUpdated, .mySegmentsUpdated, .myLargeSegmentsUpdated: if isTriggered(external: .sdkReady) { trigger(event: .sdkUpdated) - continue + return } self.triggerSdkReadyIfNeeded() @@ -166,14 +139,13 @@ class DefaultSplitEventsManager: SplitEventsManager { case .splitKilledNotification: if isTriggered(external: .sdkReady) { trigger(event: .sdkUpdated) - continue + return } case .sdkReadyTimeoutReached: if !isTriggered(external: .sdkReady) { trigger(event: SplitEvent.sdkReadyTimedOut) } } - } } // MARK: Helper functions. From 7e1738ff56dc7eb31addb752f1b37712728c25c1 Mon Sep 17 00:00:00 2001 From: Gaston Thea Date: Wed, 24 Sep 2025 12:54:36 -0300 Subject: [PATCH 2/2] Remove BlockingQueue --- Split.xcodeproj/project.pbxproj | 10 - Split/Common/Structs/BlockingQueue.swift | 95 -------- .../Collections/BlockingQueueTest.swift | 222 ------------------ 3 files changed, 327 deletions(-) delete mode 100644 Split/Common/Structs/BlockingQueue.swift delete mode 100644 SplitTests/Collections/BlockingQueueTest.swift diff --git a/Split.xcodeproj/project.pbxproj b/Split.xcodeproj/project.pbxproj index ca5cd649b..3349a1f1e 100644 --- a/Split.xcodeproj/project.pbxproj +++ b/Split.xcodeproj/project.pbxproj @@ -440,8 +440,6 @@ 9524F0042C6BFCCE00ECBF7B /* SyncSegmentsUpdateWorker.swift in Sources */ = {isa = PBXBuildFile; fileRef = 9524F0032C6BFCCE00ECBF7B /* SyncSegmentsUpdateWorker.swift */; }; 952BF5F8278DF57D00FF6181 /* TelemetryIntegrationTest.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952BF5F7278DF57D00FF6181 /* TelemetryIntegrationTest.swift */; }; 952BF5FC278DF66700FF6181 /* TelemetryConfigHelperStub.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952BF5FB278DF66700FF6181 /* TelemetryConfigHelperStub.swift */; }; - 952C2B7726447FDD009275B4 /* BlockingQueue.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952C2B7626447FDD009275B4 /* BlockingQueue.swift */; }; - 952C2B7B26449456009275B4 /* BlockingQueueTest.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952C2B7A26449456009275B4 /* BlockingQueueTest.swift */; }; 952E266B2833E35A0015D633 /* PersistentUniqueKeysStorage.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952E266A2833E35A0015D633 /* PersistentUniqueKeysStorage.swift */; }; 952E266D2833E4120015D633 /* UniqueKeyDao.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952E266C2833E4120015D633 /* UniqueKeyDao.swift */; }; 952E26702833E4F20015D633 /* UniqueKeyEntity.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952E266F2833E4F20015D633 /* UniqueKeyEntity.swift */; }; @@ -693,7 +691,6 @@ 95B02CCE28D0BDC20030EC8B /* ConcurrentDictionary.swift in Sources */ = {isa = PBXBuildFile; fileRef = 592D2CF922088606007458F9 /* ConcurrentDictionary.swift */; }; 95B02CCF28D0BDC20030EC8B /* SynchronizedDictionary.swift in Sources */ = {isa = PBXBuildFile; fileRef = 955B595328060CF100D105CD /* SynchronizedDictionary.swift */; }; 95B02CD028D0BDC20030EC8B /* KeyStorage.swift in Sources */ = {isa = PBXBuildFile; fileRef = 95118EF2281C551A00782F33 /* KeyStorage.swift */; }; - 95B02CD128D0BDC20030EC8B /* BlockingQueue.swift in Sources */ = {isa = PBXBuildFile; fileRef = 952C2B7626447FDD009275B4 /* BlockingQueue.swift */; }; 95B02CD228D0BDC20030EC8B /* LRUCache.swift in Sources */ = {isa = PBXBuildFile; fileRef = 953911A7267A4F250040433A /* LRUCache.swift */; }; 95B02CD328D0BDC20030EC8B /* Data+StringRepresentation.swift in Sources */ = {isa = PBXBuildFile; fileRef = 3B6DEE9E20EA6AE10067435E /* Data+StringRepresentation.swift */; }; 95B02CD428D0BDC20030EC8B /* Date+Utils.swift in Sources */ = {isa = PBXBuildFile; fileRef = 3B6DEE9F20EA6AE10067435E /* Date+Utils.swift */; }; @@ -1649,8 +1646,6 @@ 9524F0032C6BFCCE00ECBF7B /* SyncSegmentsUpdateWorker.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = SyncSegmentsUpdateWorker.swift; sourceTree = ""; }; 952BF5F7278DF57D00FF6181 /* TelemetryIntegrationTest.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = TelemetryIntegrationTest.swift; sourceTree = ""; }; 952BF5FB278DF66700FF6181 /* TelemetryConfigHelperStub.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = TelemetryConfigHelperStub.swift; sourceTree = ""; }; - 952C2B7626447FDD009275B4 /* BlockingQueue.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = BlockingQueue.swift; sourceTree = ""; }; - 952C2B7A26449456009275B4 /* BlockingQueueTest.swift */ = {isa = PBXFileReference; lastKnownFileType = sourcecode.swift; path = BlockingQueueTest.swift; sourceTree = ""; }; 952E266A2833E35A0015D633 /* PersistentUniqueKeysStorage.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = PersistentUniqueKeysStorage.swift; sourceTree = ""; }; 952E266C2833E4120015D633 /* UniqueKeyDao.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = UniqueKeyDao.swift; sourceTree = ""; }; 952E266F2833E4F20015D633 /* UniqueKeyEntity.swift */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.swift; path = UniqueKeyEntity.swift; sourceTree = ""; }; @@ -2625,7 +2620,6 @@ 592D2CF922088606007458F9 /* ConcurrentDictionary.swift */, 955B595328060CF100D105CD /* SynchronizedDictionary.swift */, 95118EF2281C551A00782F33 /* KeyStorage.swift */, - 952C2B7626447FDD009275B4 /* BlockingQueue.swift */, 953911A7267A4F250040433A /* LRUCache.swift */, ); path = Structs; @@ -3584,7 +3578,6 @@ 59B8B8A9220B1091003E0D5A /* SyncDictionaryCollectionWrapperTest.swift */, 59B8B8AB220B279F003E0D5A /* SyncDictionarySingleWrapperTest.swift */, 592C6AB3211B718D002D120C /* ArrayBlockingQueueTest.swift */, - 952C2B7A26449456009275B4 /* BlockingQueueTest.swift */, ); path = Collections; sourceTree = ""; @@ -4183,7 +4176,6 @@ 95B1801A2762B2C4002DC9DF /* TelemetryStatsRecorderWorker.swift in Sources */, 955B595628076BB900D105CD /* MySegmentsPayloadDecoder.swift in Sources */, 3B6DEF6020EA6AE50067435E /* BetweenMatcher.swift in Sources */, - 952C2B7726447FDD009275B4 /* BlockingQueue.swift in Sources */, 9572BA7D2AC37B7400C10FC1 /* Array+asSet.swift in Sources */, 594F5F5C253F163A00A945B4 /* BackoffCounterTimer.swift in Sources */, 3B6DEF3720EA6AE50067435E /* SplitManager.swift in Sources */, @@ -4762,7 +4754,6 @@ 596A7B71235729B5004F6139 /* DestroyTest.swift in Sources */, 95B180222763C00C002DC9DF /* HttpTelemetryStatsRecorderStub.swift in Sources */, 595AD25924E59E6200A7B750 /* EventStreamParserTest.swift in Sources */, - 952C2B7B26449456009275B4 /* BlockingQueueTest.swift in Sources */, C526DE4C2D9B09EB0051DAB8 /* ImpressionsPropertiesE2ETest.swift in Sources */, 9572BA812AC4A4B100C10FC1 /* SplitChangeProcessorTests.swift in Sources */, 95ABF4CA29352589006ED016 /* AnyValueValidatorStub.swift in Sources */, @@ -4908,7 +4899,6 @@ 9500D9D12C61814000383593 /* MySegmentEntity.swift in Sources */, 95B02CCF28D0BDC20030EC8B /* SynchronizedDictionary.swift in Sources */, 95B02CD028D0BDC20030EC8B /* KeyStorage.swift in Sources */, - 95B02CD128D0BDC20030EC8B /* BlockingQueue.swift in Sources */, 95B02CD228D0BDC20030EC8B /* LRUCache.swift in Sources */, 95B02CD328D0BDC20030EC8B /* Data+StringRepresentation.swift in Sources */, 95B02CD428D0BDC20030EC8B /* Date+Utils.swift in Sources */, diff --git a/Split/Common/Structs/BlockingQueue.swift b/Split/Common/Structs/BlockingQueue.swift deleted file mode 100644 index c8b750940..000000000 --- a/Split/Common/Structs/BlockingQueue.swift +++ /dev/null @@ -1,95 +0,0 @@ -// -// BlockingQueue.swift -// Split -// -// Created by Javier Avrudsky on 06/05/2021. -// Copyright © 2021 Split. All rights reserved. -// - -import Foundation - -enum BlockingQueueError: Error { - case hasBeenStopped - case noElementAvailable -} - -class GenericBlockingQueue { - private var elements: [Item] - private let dispatchQueue: DispatchQueue - private let semaphore: DispatchSemaphore - private var isStopped = false - init() { - dispatchQueue = DispatchQueue(label: "split-generic-blocking-queue", - attributes: .concurrent) - semaphore = DispatchSemaphore(value: 0) - elements = [Item]() - } - - func add(_ item: Item) { - dispatchQueue.async(flags: .barrier) { [weak self] in - if let self = self { - if self.isStopped { return } - self.elements.append(item) - self.semaphore.signal() - } - } - } - - func take() throws -> Item { - var item: Item? - // Checks if stopped before waiting - try checkIfStopped() - self.semaphore.wait(timeout: .now() + .milliseconds(100)) - try dispatchQueue.sync(flags: .barrier) { - // Checks if thread was awaked by stop or interruption - try checkIfStopped() - if elements.count > 0 { - item = elements.removeFirst() - } - } - guard let foundItem = item else { - throw BlockingQueueError.noElementAvailable - } - return foundItem - } - - func stop() { - dispatchQueue.async(flags: .barrier) { [weak self] in - if let self = self { - self.isStopped = true - self.elements.removeAll() - self.semaphore.signal() - } - } - } - - // Use this function within the queue - private func checkIfStopped() throws { - if self.isStopped { - throw BlockingQueueError.hasBeenStopped - } - } -} - -// Protocol to allow mocking -protocol InternalEventBlockingQueue { - func add(_ item: SplitInternalEvent) - func take() throws -> SplitInternalEvent - func stop() -} - -class DefaultInternalEventBlockingQueue: InternalEventBlockingQueue { - let blockingQueue = GenericBlockingQueue() - func add(_ item: SplitInternalEvent) { - blockingQueue.add(item) - } - - func take() throws -> SplitInternalEvent { - let value = try blockingQueue.take() - return value - } - - func stop() { - blockingQueue.stop() - } -} diff --git a/SplitTests/Collections/BlockingQueueTest.swift b/SplitTests/Collections/BlockingQueueTest.swift deleted file mode 100644 index 03ad78675..000000000 --- a/SplitTests/Collections/BlockingQueueTest.swift +++ /dev/null @@ -1,222 +0,0 @@ -// -// BlockingQueueTest.swift -// SplitTests -// -// Created by Javier Avrudsky on 06/05/2021. -// Copyright © 2021 Split. All rights reserved. -// - -import Foundation -import XCTest -@testable import Split - -class BlockingQueueTest: XCTestCase { - - override func setUp() { - } - - func testAddAndTake() { - let endExp = XCTestExpectation() - var local = [SplitInternalEvent]() - let globalQ = DispatchQueue.test - let queue = DefaultInternalEventBlockingQueue() - - globalQ.async { - while true { - do { - let event = try queue.take() - local.append(event) - if local.count == 4 { - endExp.fulfill() - } - } catch { - } - } - } - globalQ.asyncAfter(deadline: .now() + 1) { - queue.add(SplitInternalEvent.mySegmentsLoadedFromCache) - globalQ.asyncAfter(deadline: .now() + 1) { - queue.add(SplitInternalEvent.splitsLoadedFromCache) - } - } - - queue.add(SplitInternalEvent.splitsUpdated) - queue.add(SplitInternalEvent.mySegmentsUpdated) - wait(for: [endExp], timeout: 10) - XCTAssertEqual(SplitInternalEvent.splitsUpdated, local[0]) - XCTAssertEqual(SplitInternalEvent.mySegmentsUpdated, local[1]) - XCTAssertEqual(SplitInternalEvent.mySegmentsLoadedFromCache, local[2]) - XCTAssertEqual(SplitInternalEvent.splitsLoadedFromCache, local[3]) - } - - func testInterrupt() { - let endExp = XCTestExpectation() - var local = [SplitInternalEvent]() - let globalQ = DispatchQueue.global() - let queue = DefaultInternalEventBlockingQueue() - var interrupted = false - - globalQ.async { - while true { - do { - let event = try queue.take() - local.append(event) - } catch BlockingQueueError.noElementAvailable { - continue - } catch { - endExp.fulfill() - interrupted = true - } - } - } - globalQ.asyncAfter(deadline: .now() + 1) { - queue.add(SplitInternalEvent.mySegmentsLoadedFromCache) - globalQ.asyncAfter(deadline: .now() + 1) { - queue.stop() - } - } - - queue.add(SplitInternalEvent.splitsUpdated) - queue.add(SplitInternalEvent.mySegmentsUpdated) - - wait(for: [endExp], timeout: 10) - XCTAssertEqual(SplitInternalEvent.splitsUpdated, local[0]) - XCTAssertEqual(SplitInternalEvent.mySegmentsUpdated, local[1]) - XCTAssertEqual(SplitInternalEvent.mySegmentsLoadedFromCache, local[2]) - XCTAssertTrue(interrupted) - } - - func stressAddAndTakeTest() { - let endExp = XCTestExpectation() - var local = [SplitInternalEvent]() - let globalQ = DispatchQueue.global() - let qu1 = DispatchQueue(label: "stress-test-take-q1", attributes: .concurrent) - let qu2 = DispatchQueue(label: "stress-test-take-q2", attributes: .concurrent) - let qu3 = DispatchQueue(label: "stress-test-take-q3", attributes: .concurrent) - let qu4 = DispatchQueue(label: "stress-test-take-q4", attributes: .concurrent) -// let qu5 = DispatchQueue(label: "stress-test-take-q4", attributes: .concurrent) - - let quA1 = DispatchQueue(label: "stress-test-take-qa1", attributes: .concurrent) - let quA2 = DispatchQueue(label: "stress-test-take-qa2", attributes: .concurrent) - let quA3 = DispatchQueue(label: "stress-test-take-qa3", attributes: .concurrent) - - let stopq = DispatchQueue(label: "stress-test-take-stop", attributes: .concurrent) - let queue = DefaultInternalEventBlockingQueue() - - globalQ.async { - for _ in 0..<50000 { - do { - let event = try queue.take() - local.append(event) - print("Took: \(event)") - } catch { - } - } - - } - - quA1.async { - for _ in 0..<50000 { - do { - let event = try queue.take() - local.append(event) - print("Took QA1: \(event)") - } catch { - print("\n\n\nERROR!!!!: \(error) \n\n\n") - } - } - } - - quA2.async { - for _ in 0..<50000 { - do { - let event = try queue.take() - local.append(event) - Thread.sleep(forTimeInterval: 0.3) - print("Took QA2: \(event)") - } catch { - } - } - } - - quA3.async { - for _ in 0..<50000 { - do { - Thread.sleep(forTimeInterval: 0.5) - let event = try queue.take() - local.append(event) - print("Took QA3: \(event)") - } catch { - } - } - } - - qu1.async { - for _ in 1..<100000 { - queue.add(SplitInternalEvent.splitsUpdated) - print("qu1 add") - Thread.sleep(forTimeInterval: 0.2) - } - } - - qu2.async { - for _ in 1..<10000 { - print("qu2 add") - queue.add(SplitInternalEvent.sdkReadyTimeoutReached) - Thread.sleep(forTimeInterval: 0.5) - } - } - - qu3.async { - for _ in 1..<10000 { - print("qu3 add") - queue.add(SplitInternalEvent.splitsUpdated) - Thread.sleep(forTimeInterval: 0.8) - } - } - - qu4.async { - for _ in 1..<10000 { - print("qu4 add") - queue.add(SplitInternalEvent.mySegmentsUpdated) - sleep(1) - } - } - - stopq.asyncAfter(deadline: .now() + 10) { - print("\n\n\n\n\n\n**************** STOP 10 \n\n\n\n\n") - queue.stop() - } - - stopq.asyncAfter(deadline: .now() + 20) { - print("\n\n\n\n\n\n**************** STOP 20 \n\n\n\n\n") - queue.stop() - } - - stopq.async { - for i in 1..<10000 { - Thread.sleep(forTimeInterval: 2) - print("\n\n\n\n\n\n**************** STOP i=\(i) \n\n\n\n\n") - queue.stop() - } - } - - wait(for: [endExp], timeout: 50) - } - - func testCallingTakeOnEmptyQueueThrowsNoElementAvailable() { - let endExp = XCTestExpectation() - let queue = DefaultInternalEventBlockingQueue() - - do { - _ = try queue.take() - } catch BlockingQueueError.noElementAvailable { - endExp.fulfill() - } catch { - XCTFail() - } - - self.wait(for: [endExp], timeout: 1) - } -} -