Skip to content

Commit

Permalink
Allow user to specify enqueue DispatchQueue to fix multi-thread enque…
Browse files Browse the repository at this point in the history
…ue crash (#403)
  • Loading branch information
Lucas Nelaupe authored May 5, 2022
1 parent ab9f88f commit 16a15e3
Show file tree
Hide file tree
Showing 3 changed files with 48 additions and 2 deletions.
1 change: 0 additions & 1 deletion Sources/SwiftQueue/JobBuilder.swift
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,6 @@ public final class JobBuilder {
assert(JSONSerialization.isValidJSONObject(info.params))
info.constraints.append(PersisterConstraint(serializer: manager.params.serializer, persister: manager.params.persister))
}

manager.enqueue(info: info)
}

Expand Down
24 changes: 23 additions & 1 deletion Sources/SwiftQueue/SwiftQueueManager.swift
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,16 @@ public final class SwiftQueueManager {
/// Schedule a job to the queue
/// TODO Need to remove this method
public func enqueue(info: JobInfo) {
if let thread = params.enqueueThread {
thread.sync(flags: .barrier) {
enqueueInMainThread(info: info)
}
} else {
enqueueInMainThread(info: info)
}
}

private func enqueueInMainThread(info: JobInfo) {
let queue = getQueue(queueName: info.queueName)
let job = queue.createHandler(type: info.type, params: info.params)
let constraints = info.constraints
Expand Down Expand Up @@ -167,14 +177,17 @@ internal struct SqManagerParams {

var initInBackground: Bool

var enqueueThread: DispatchQueue?

init(jobCreator: JobCreator,
queueCreator: QueueCreator,
persister: JobPersister = UserDefaultsPersister(),
serializer: JobInfoSerializer = DecodableSerializer(maker: DefaultConstraintMaker()),
logger: SwiftQueueLogger = NoLogger.shared,
listener: JobListener? = nil,
initInBackground: Bool = false,
dispatchQueue: DispatchQueue = DispatchQueue.global(qos: DispatchQoS.QoSClass.utility)
dispatchQueue: DispatchQueue = DispatchQueue.global(qos: DispatchQoS.QoSClass.utility),
enqueueThread: DispatchQueue? = nil
) {
self.jobCreator = jobCreator
self.queueCreator = queueCreator
Expand All @@ -184,6 +197,7 @@ internal struct SqManagerParams {
self.listener = listener
self.initInBackground = initInBackground
self.dispatchQueue = dispatchQueue
self.enqueueThread = enqueueThread
}

}
Expand Down Expand Up @@ -242,6 +256,14 @@ public final class SwiftQueueManagerBuilder {
return self
}

/// Use a single DispatchQueue to enqueue jobs
/// This can solve crashes when calling 'enqueue' in a multi-thread environment
public func set(enqueueDispatcher: DispatchQueue) -> Self {
params.enqueueThread = enqueueDispatcher
return self
}


/// Get an instance of `SwiftQueueManager`
public func build() -> SwiftQueueManager {
return SwiftQueueManager(params: params, isSuspended: isSuspended)
Expand Down
25 changes: 25 additions & 0 deletions Tests/SwiftQueueTests/SwiftQueueManagerTests.swift
Original file line number Diff line number Diff line change
Expand Up @@ -305,4 +305,29 @@ class SwiftQueueManagerTests: XCTestCase {
job.assertError(queueError: .canceled)
}

func testConcurrentScheduling() {
let (type, job) = (UUID().uuidString, TestJob())
let creator = TestCreator([type: job])
let persister = PersisterTracker(key: UUID().uuidString)
let manager = SwiftQueueManagerBuilder(creator: creator)
.set(isSuspended: true)
.set(enqueueDispatcher: .main)
.set(persister: persister).build()


let concurrentQueue = DispatchQueue(label: "com.test.concurrent", attributes: .concurrent)
for _ in 0..<10 {
concurrentQueue.async {
JobBuilder(type: type).parallel(queueName: UUID().uuidString).schedule(manager: manager)
}
}

for _ in 0..<10 {
DispatchQueue(label: "com.test.concurrent", attributes: .concurrent).async {
JobBuilder(type: type).parallel(queueName: UUID().uuidString).schedule(manager: manager)
}
}

}

}

0 comments on commit 16a15e3

Please sign in to comment.