• 1

• 271
• 3
• 10

• 271
• 3
• 10

I would like to develop a kind of template or canonical implementation for a concurrent subclass of NSOperation in Swift.

See here for my previous request which is implemented in Objective-C. In that previous solution, there are still issues when the operation will be cancelled while it has not yet been started. There is also more or less sloppy approach regarding concurrency when accessing the properties isExecuting and isFinished.

In this new implementation, using Swift, I also try to address the previous issues.

I appreciate any comments, suggestions and questions.

Requirements:

• Thread safe API.
• start shall only start the task once, otherwise it should have no effect.
• The operation object should not prematurely deinit while the task is executing.
• Provide a reusable subclass of NSOperation that can be easily used as a base class of any custom operation.

References:

https://developer.apple.com/library/mac/documentation/Cocoa/Reference/NSOperation_class/Reference/Reference.html

Generic class Operation<T>:

import Foundation
public enum OperationError : Int, ErrorType {
 case Cancelled = -1
 case NoTask = -1000
}
private var queue_ID_key = 0
public class Operation<T> : NSOperation {
 
 public typealias ValueType = T
 public typealias ResultType = Result<ValueType>
 public typealias ContinuationFunc = ResultType -> ()
 private var _result: Result<ValueType>?
 private let _syncQueue = dispatch_queue_create("operation.sync_queue", DISPATCH_QUEUE_SERIAL);
 private var _isExecuting = false
 private var _isFinished = false
 private var _continuation: ContinuationFunc?
 private var _self: NSOperation? = nil
 public required init(continuation: ContinuationFunc? = nil) {
 let context = UnsafeMutablePointer<Void>(Unmanaged<dispatch_queue_t>.passUnretained(_syncQueue).toOpaque())
 dispatch_queue_set_specific(_syncQueue, &queue_ID_key, context, nil)
 _continuation = continuation
 }
 
 /**
 Override in subclass
 `task` will be called within the synchronization context. The completion handler
 should be called on a private execution context.
 */
 public func task(completion: ResultType -> ()) {
 assert(true, "override in subclass")
 dispatch_async(dispatch_get_global_queue(0, 0)) {
 completion(Result(error: OperationError.NoTask))
 }
 }
 
 
 public final override var asynchronous : Bool { return true }
 
 public final override var executing : Bool {
 get {
 // Note: `executing` will be called recursively from `willChangeValueForKey` and `didChangeValueForKey`. Thus, we need a counter measurement to prevent a deadlock:
 if is_synchronized() {
 return _isExecuting
 }
 var result = false
 dispatch_sync(_syncQueue) {
 result = self._isExecuting
 }
 return result
 }
 set {
 dispatch_async(_syncQueue) {
 if self._isExecuting != newValue {
 self.willChangeValueForKey("isExecuting")
 self._isExecuting = newValue
 self.didChangeValueForKey("isExecuting")
 }
 }
 }
 }
 
 public final override var finished : Bool {
 get {
 // Note: `finished` will be called recursively from `willChangeValueForKey` and `didChangeValueForKey`. Thus, we need a counter measurement to prevent a deadlock:
 if is_synchronized() {
 return _isFinished
 }
 var result = false
 dispatch_sync(_syncQueue) {
 result = self._isFinished
 }
 return result
 }
 set {
 dispatch_async(_syncQueue) {
 if self._isFinished != newValue {
 self.willChangeValueForKey("isFinished")
 self._isExecuting = newValue
 self.didChangeValueForKey("isFinished")
 }
 }
 }
 }
 
 
 /**
 Returns the result of the operation.
 */
 public final var result: ResultType? {
 var result: ResultType? = nil
 dispatch_sync(_syncQueue) {
 result = self._result;
 }
 return result;
 }
 
 
 /**
 Starts the operation.
 `start` has no effect when the operation has been cancelled, or when it
 has already been started.
 */
 public final override func start() {
 dispatch_async(_syncQueue) {
 if !self.cancelled && !self._isFinished && !self._isExecuting {
 self.executing = true;
 self._self = self; // make self immortal for the duration of the task
 self.task() { result in
 dispatch_async(self._syncQueue) {
 self._result = result
 self.terminate()
 }
 }
 }
 }
 }
 
 /**
 `terminate` will be called when `self` has been cancelled or when `self`'s task
 completes. Sets its `finished` property to `true`.
 If there is a continuation it will be called. If self has a `completionBlock`
 it will be called, too.
 */
 private final func terminate() {
 assert(_result != nil)
 if _isExecuting {
 self.executing = false
 }
 self.finished = true
 if let continuation = _continuation {
 _continuation = nil
 let result = _result
 dispatch_async(dispatch_get_global_queue(0, 0)) {
 continuation(result!)
 }
 }
 _self = nil
 }
 
 /**
 Asynchronously requests a cancellation.
 If the operation has been cancelled already or if it is finished, the
 method has no effect. Otherwise, calls `super.cancel()`, sets the result to
 `OperationError.Cancelled` and sets the `finished` property to `true`. If the 
 operation is executing it sets the `executing` property to `false`. If there
 is a continuation, it will be called with the operation's result as its
 argument.
 */
 public final override func cancel() {
 dispatch_async(_syncQueue) {
 // Cancel should not have an effect if self is already cancelled, or self has been finished.
 // We test `_result` to accomplish this:
 if self._result == nil {
 self._result = Result(error: OperationError.Cancelled)
 super.cancel()
 self.terminate()
 }
 }
 }
 
 /**
 Returns `true` if the current execution context is synchronized with
 the `_syncQueue`.
 */
 private final func is_synchronized() -> Bool {
 let context = UnsafeMutablePointer<Void>(Unmanaged<dispatch_queue_t>.passUnretained(_syncQueue).toOpaque())
 return dispatch_get_specific(&queue_ID_key) == context
 }
 
 
}

(The implementation for class Result<T> is not shown)

Sample Custom Class:

public class MyOperation : Operation<String> { 
 private var _duration = 1.0 
 
 public required init(continuation: ContinuationFunc? = nil) {
 super.init(continuation: continuation)
 }
 
 public init(duration: Double, continuation: ContinuationFunc? = nil) {
 _duration = duration
 super.init(continuation: continuation)
 }
 
 public override func task(completion: ResultType -> ()) {
 let delay: Double = _duration
 let t = dispatch_time(DISPATCH_TIME_NOW, Int64(delay * Double(NSEC_PER_SEC)))
 dispatch_after(t, dispatch_get_global_queue(0, 0)) {
 completion(Result("OK"))
 }
 }
 
}

Sample usage:

func testOperationShoudlCompleteWithSuccess() {
 let expect1 = self.expectationWithDescription("continuation1 should be called")
 let op = MyOperation() { result in
 expect1.fulfill()
 }
 op.start()
 waitForExpectationsWithTimeout(2, handler: nil)
}
func testStartedOperationShoudlCompleteWithCancellationError() {
 let expect1 = self.expectationWithDescription("continuation1 should be called")
 let op = MyOperation() { result in
 do {
 let _ = try result.value()
 XCTFail("unexpected success")
 }
 catch let error {
 XCTAssertTrue(error is OperationError)
 print("Error: \(error)")
 }
 expect1.fulfill()
 }
 op.start()
 op.cancel()
 
 waitForExpectationsWithTimeout(2, handler: nil)
}
func testScheduledOperationsShouldCompleteWithSuccess() {
 
 let expectations = [
 self.expectationWithDescription("continuation1 should be called"),
 self.expectationWithDescription("continuation2 should be called"),
 self.expectationWithDescription("continuation3 should be called"),
 ]
 
 let operations = [
 MyOperation(duration: 0.1) { result in
 XCTAssertTrue(result.isSuccess)
 expectations[0].fulfill()
 },
 MyOperation(duration: 0.1) { result in
 XCTAssertTrue(result.isSuccess)
 expectations[1].fulfill()
 },
 MyOperation(duration: 0.1) { result in
 XCTAssertTrue(result.isSuccess)
 expectations[2].fulfill()
 } 
 ]
 let queue = NSOperationQueue()
 queue.addOperations(operations, waitUntilFinished: false)
 
 waitForExpectationsWithTimeout(0.4, handler: nil)
}