Merge pull request swiftlang#1260 from pushkarnk/urlsession-fixes-4.0

Author: swift-ci

Foundation/URLSession/URLSession.swift

@@ -172,11 +172,13 @@ import CoreFoundation
 import Dispatch
 
 
+fileprivate let globalVarSyncQ = DispatchQueue(label: "org.swift.Foundation.URLSession.GlobalVarSyncQ")
 fileprivate var sessionCounter = Int32(0)
 fileprivate func nextSessionIdentifier() -> Int32 {
-    //TODO: find an alternative for OSAtomicIncrement32Barrier() on Linux
-    sessionCounter += 1
-    return sessionCounter
+    return globalVarSyncQ.sync {
+        sessionCounter += 1
+        return sessionCounter
+    }
 }
 public let NSURLSessionTransferSizeUnknown: Int64 = -1
 
@@ -397,9 +399,11 @@ extension URLSession._Request {
 
 fileprivate extension URLSession {
     func createNextTaskIdentifier() -> Int {
-        let i = nextTaskIdentifier
-        nextTaskIdentifier += 1
-        return i
+        return workQueue.sync {
+            let i = nextTaskIdentifier
+            nextTaskIdentifier += 1
+            return i
+        }
     }
 }
 

Foundation/URLSession/URLSessionTask.swift

@@ -30,13 +30,10 @@ open class URLSessionTask : NSObject, NSCopying {
     internal var session: URLSessionProtocol! //change to nil when task completes
     internal let body: _Body
     fileprivate var _protocol: URLProtocol? = nil
-
+    private let syncQ = DispatchQueue(label: "org.swift.URLSessionTask.SyncQ")
+    
     /// All operations must run on this queue.
     internal let workQueue: DispatchQueue 
-    /// Using dispatch semaphore to make public attributes thread safe.
-    /// A semaphore is a simpler option against the usage of concurrent queue
-    /// as the critical sections are very short.
-    fileprivate let semaphore = DispatchSemaphore(value: 1)    
 
     public override init() {
         // Darwin Foundation oddly allows calling this initializer, even though
@@ -62,7 +59,8 @@ open class URLSessionTask : NSObject, NSCopying {
     }
     internal init(session: URLSession, request: URLRequest, taskIdentifier: Int, body: _Body) {
         self.session = session
-        self.workQueue = session.workQueue
+        /* make sure we're actually having a serial queue as it's used for synchronization */
+        self.workQueue = DispatchQueue.init(label: "org.swift.URLSessionTask.WorkQueue", target: session.workQueue)
         self.taskIdentifier = taskIdentifier
         self.originalRequest = request
         self.body = body
@@ -108,31 +106,19 @@ open class URLSessionTask : NSObject, NSCopying {
     /// May differ from originalRequest due to http server redirection
     /*@NSCopying*/ open internal(set) var currentRequest: URLRequest? {
         get {
-            semaphore.wait()
-            defer {
-                semaphore.signal()
-            }
-            return self._currentRequest
+            return self.syncQ.sync { return self._currentRequest }
         }
         set {
-            semaphore.wait()
-            self._currentRequest = newValue
-            semaphore.signal()
+            self.syncQ.sync { self._currentRequest = newValue }
         }
     }
     fileprivate var _currentRequest: URLRequest? = nil
     /*@NSCopying*/ open internal(set) var response: URLResponse? {
         get {
-            semaphore.wait()
-            defer {
-                semaphore.signal()
-            }
-            return self._response
+            return self.syncQ.sync { return self._response }
         }
         set {
-            semaphore.wait()
-            self._response = newValue
-            semaphore.signal()
+            self.syncQ.sync { self._response = newValue }
         }
     }
     fileprivate var _response: URLResponse? = nil
@@ -145,33 +131,21 @@ open class URLSessionTask : NSObject, NSCopying {
     /// Number of body bytes already received
     open internal(set) var countOfBytesReceived: Int64 {
         get {
-            semaphore.wait()
-            defer {
-                semaphore.signal()
-            }
-            return self._countOfBytesReceived
+            return self.syncQ.sync { return self._countOfBytesReceived }
         }
         set {
-            semaphore.wait()
-            self._countOfBytesReceived = newValue
-            semaphore.signal()
+            self.syncQ.sync { self._countOfBytesReceived = newValue }
         }
     }
     fileprivate var _countOfBytesReceived: Int64 = 0
 
     /// Number of body bytes already sent */
     open internal(set) var countOfBytesSent: Int64 {
         get {
-            semaphore.wait()
-            defer {
-                semaphore.signal()
-            }
-            return self._countOfBytesSent
+            return self.syncQ.sync { return self._countOfBytesSent }
         }
         set {
-            semaphore.wait()
-            self._countOfBytesSent = newValue
-            semaphore.signal()
+            self.syncQ.sync { self._countOfBytesSent = newValue }
         }
     }
 
@@ -211,16 +185,10 @@ open class URLSessionTask : NSObject, NSCopying {
      */
     open var state: URLSessionTask.State {
         get {
-            semaphore.wait()
-            defer {
-                semaphore.signal()
-            }
-            return self._state
+            return self.syncQ.sync { self._state }
         }
         set {
-            semaphore.wait()
-            self._state = newValue
-            semaphore.signal()
+            self.syncQ.sync { self._state = newValue }
         }
     }
     fileprivate var _state: URLSessionTask.State = .suspended
@@ -315,16 +283,10 @@ open class URLSessionTask : NSObject, NSCopying {
     /// URLSessionTask.highPriority, but use is not restricted to these.
     open var priority: Float {
         get {
-            semaphore.wait()
-            defer {
-                semaphore.signal()
-            }
-            return self._priority
+            return self.workQueue.sync { return self._priority }
         }
         set {
-            semaphore.wait()
-            self._priority = newValue
-            semaphore.signal()
+            self.workQueue.sync { self._priority = newValue }
         }
     }
     fileprivate var _priority: Float = URLSessionTask.defaultPriority
@@ -569,7 +531,9 @@ extension _ProtocolClient : URLProtocolClient {
             session.delegateQueue.addOperation {
                 delegate.urlSession(session, task: task, didCompleteWithError: nil)
                 task.state = .completed
-                session.taskRegistry.remove(task)
+                task.workQueue.async {
+                    session.taskRegistry.remove(task)
+                }
             }
         case .noDelegate:
             task.state = .completed
@@ -625,7 +589,9 @@ extension _ProtocolClient : URLProtocolClient {
             session.delegateQueue.addOperation {
                 delegate.urlSession(session, task: task, didCompleteWithError: error as Error)
                 task.state = .completed
-                session.taskRegistry.remove(task)
+                task.workQueue.async {
+                    session.taskRegistry.remove(task)
+                }
             }
         case .noDelegate:
             task.state = .completed
@@ -634,7 +600,9 @@ extension _ProtocolClient : URLProtocolClient {
             session.delegateQueue.addOperation {
                 completion(nil, nil, error)
                 task.state = .completed
-                session.taskRegistry.remove(task)
+                task.workQueue.async {
+                    session.taskRegistry.remove(task)
+                }
             }
         case .downloadCompletionHandler(let completion):
             session.delegateQueue.addOperation {

Foundation/URLSession/http/MultiHandle.swift

@@ -39,11 +39,7 @@ extension URLSession {
         let group = DispatchGroup()
         fileprivate var easyHandles: [_EasyHandle] = []
         fileprivate var timeoutSource: _TimeoutSource? = nil
-
-        //SR-4567: we need to synchronize the register/unregister commands to the epoll machinery in libdispatch
-        fileprivate let commandQueue: DispatchQueue = DispatchQueue(label: "Register-unregister synchronization")
-        fileprivate var cancelInProgress: DispatchSemaphore? = nil
-
+        
         init(configuration: URLSession._Configuration, workQueue: DispatchQueue) {
             queue = DispatchQueue(label: "MultiHandle.isolation", target: workQueue)
             setupCallbacks()
@@ -103,33 +99,25 @@ fileprivate extension URLSession._MultiHandle {
         // through libdispatch (DispatchSource) and store the source(s) inside
         // a `SocketSources` which we in turn store inside libcurl's multi handle
         // by means of curl_multi_assign() -- we retain the object fist.
-        commandQueue.async {
-            self.cancelInProgress?.wait()
-            self.cancelInProgress = nil
-
-            let action = _SocketRegisterAction(rawValue: CFURLSessionPoll(value: what))
-            var socketSources = _SocketSources.from(socketSourcePtr: socketSourcePtr)
-            if socketSources == nil && action.needsSource {
-                let s = _SocketSources()
-                let p = Unmanaged.passRetained(s).toOpaque()
-                CFURLSessionMultiHandleAssign(self.rawHandle, socket, UnsafeMutableRawPointer(p))
-                socketSources = s
-            } else if socketSources != nil && action == .unregister {
-                //the beginning of an unregister operation
-                self.cancelInProgress = DispatchSemaphore(value: 0)
-                // We need to release the stored pointer:
-                if let opaque = socketSourcePtr {
-                    Unmanaged<_SocketSources>.fromOpaque(opaque).release()
-                }
-                socketSources?.tearDown(self.cancelInProgress)
-                socketSources = nil
+        let action = _SocketRegisterAction(rawValue: CFURLSessionPoll(value: what))
+        var socketSources = _SocketSources.from(socketSourcePtr: socketSourcePtr)
+        if socketSources == nil && action.needsSource {
+            let s = _SocketSources()
+            let p = Unmanaged.passRetained(s).toOpaque()
+            CFURLSessionMultiHandleAssign(rawHandle, socket, UnsafeMutableRawPointer(p))
+            socketSources = s
+        } else if socketSources != nil && action == .unregister {
+            // We need to release the stored pointer:
+            if let opaque = socketSourcePtr {
+                Unmanaged<_SocketSources>.fromOpaque(opaque).release()
             }
-            if let ss = socketSources {
-                let handler = DispatchWorkItem { [weak self] in
-                    self?.performAction(for: socket)
-                }
-                ss.createSources(with: action, fileDescriptor: Int(socket), queue: self.queue, handler: handler)
+            socketSources = nil
+        }
+        if let ss = socketSources {
+            let handler = DispatchWorkItem { [weak self] in
+                self?.performAction(for: socket)
             }
+            ss.createSources(with: action, fileDescriptor: Int(socket), queue: queue, handler: handler)
         }
         return 0
     }
@@ -411,18 +399,12 @@ fileprivate class _SocketSources {
         s.resume()
     }
 
-    func tearDown(_ cancelInProgress: DispatchSemaphore?) {
-        let cancelHandler = DispatchWorkItem {
-            //the real end of an unregister operation!
-            cancelInProgress?.signal()
-        }
+    func tearDown() {
         if let s = readSource {
-            s.setCancelHandler(handler: cancelHandler)
             s.cancel()
         }
         readSource = nil
         if let s = writeSource {
-            s.setCancelHandler(handler: cancelHandler)
             s.cancel()
         }
         writeSource = nil

TestFoundation/HTTPServer.swift

@@ -26,6 +26,8 @@ import Dispatch
 #endif
 
 public let globalDispatchQueue = DispatchQueue.global()
+public let dispatchQueueMake: (String) -> DispatchQueue = { DispatchQueue.init(label: $0) }
+public let dispatchGroupMake: () -> DispatchGroup = DispatchGroup.init
 
 struct _HTTPUtils {
     static let CRLF = "\r\n"