Mac上XPC多进程通讯的完整解决方案

文档更新说明

最后更新 2021年4月1日
首次更新 2021年4月1日

前言

本文主要讲述如何在Mac上利用XPC技术实现多进程通讯，包括全局双向通讯，合法性校验，通讯协议版本校验，以及多线程注意事项等四部分。

经过半个月的开发，目前已经开发完相关功能，包括上面几点，本文会讲述每一个部分的技术细节，相信能给桌面开发者一些灵感和提示。

忘了说了，这篇本来是3月份的文章的，应该工作原因涉及到的技术一直没经过检验，所以拖到现在才发布的：）

双向通讯

XPC技术本身就支持双向通讯的，不过有别于IP/TCP协议，XPC被苹果封装到了对象级别，两个进程可以相互调用实现已经设定好的高级语言中的协议Protocol，这部分属于API级别的，下面简单演示。

通讯原理图：

上面就是利用XPC进行全局多进程双向通讯的结构图，一共有三个XPC服务，分别是主程序里，服务程序里，中介XPC里。

左边是为了获取服务程序的匿名端口，主程序拿到匿名端口之后，就可以直接通过XPC与服务程序通讯了。这里XPC的接口使用方法，可以参考这一篇文章：ObjC 中国 - XPC使用方法

XPC有两种类型，私有XPC服务和全局XPC服务，私有的XPC服务存放在App包内，只能由APP使用，而且一个APP对应地启动一个XPC进程，属于一对一。

全局XPC服务，则必须由launchd进程启动，启动方法参考之前的文章进程启动原理

上图里的中介XPC其实就是一个全局XPC，主程序通过请求中介XPC的接口，让中介XPC发出全局通知，服务程序接收到通知后创建匿名端口，传给中介XPC，中介XPC获取到匿名端口之后返回给主程序，这样主程序就可以拿到服务程序的匿名端口，进而同服务程序创建连接进行通讯。

这就是全局通讯原理，全局是指任意两个进程都可以通过XPC建立连接的意思。

下面附上中介XPC的服务代码，一共是2个接口，查询端口，提供给主程序调用；另一个是存储接口，提供给服务进程存放匿名端口。

import Cocoa

class AgentXPCServiceObject: AgentXPCServiceProtocol {
    static let instance = AgentXPCServiceObject()

    // 同一个进程发起XPC请求时, 所有请求都是串行发出的, 需要支持多客户端连接可以加锁.
    var endpoints: [String: NSXPCListenerEndpoint] = [:]
    var waiterMap: [String: DispatchGroup] = [:]
    var waiterFlags: [String: Bool] = [:]
    var count = 0

    func getWaiter(tag: String) -> DispatchGroup {
        var waiter = waiterMap[tag]
        if waiter == nil {
            waiter = DispatchGroup()
            waiterMap[tag] = waiter
        }
        return waiter!
    }

    func setWaiter(tag: String, w: DispatchGroup?) {
        waiterMap[tag] = w
    }

    func getEndpoint(tag: String) -> NSXPCListenerEndpoint? {
        return endpoints[tag]
    }

    func setEndpoint(tag: String, endpoint: NSXPCListenerEndpoint?) {
        endpoints[tag] = endpoint
    }

    func getWaiterFlag(tag: String) -> Bool {
        let f = waiterFlags[tag]
        if f == nil {
            waiterFlags[tag] = false
            return false
        }
        return f!
    }

    func setWaiterFlag(tag: String, f: Bool) {
        waiterFlags[tag] = f
    }

    /// 获取指定tag对应的匿名端口
    /// - Parameters:
    ///   - tag: tag, 预留的多用户参数, 暂时无效, 切记, 同一个tag不支持并发获取endpoint.
    ///   - reply: 回调
    func queryEndPoint(tag: String, withReply reply: (NSXPCListenerEndpoint?, NSError?) -> Void) {

        let defaultTag = "tag"

        // 清空老的endpoint, 确保每次查询到的endpoint都是主程序重新下发的
        setEndpoint(tag: defaultTag, endpoint: nil)

        // 等待DACS推送匿名端口, 超时返回错误
        setWaiterFlag(tag: defaultTag, f: true)
        let waiter = getWaiter(tag: defaultTag)
        waiter.enter()

        // 推送全局通知, 通知dacs主程序post端口
        let nfcenter = CFNotificationCenterGetDistributedCenter()
        // 如果有需要, 可以在全局通知里附带tag信息, 这样dacs接受到的时候就知道给哪个tag发送端口, 也就可以支持多端口了.
        CFNotificationCenterPostNotification(nfcenter, CFNotificationName("com.DACSAgentXPC.query.endpoint" as CFString), nil, nil, false)

        let status = waiter.wait(timeout: DispatchTime.now() + .seconds(6))

        var ep: NSXPCListenerEndpoint?
        if status == .success {
            ep = getEndpoint(tag: defaultTag)
        }

        if ep != nil {
            reply(ep!, nil)
            return
        } else {
            reply(nil, NSError(domain: "can't get endpoint from server", code: -1, userInfo: nil))
            // 撤销group的标记
            setWaiterFlag(tag: defaultTag, f: false)
            waiter.leave()
            // 清理waiter
            setWaiter(tag: defaultTag, w: nil)
        }
    }

    func postEndPoint(tag: String, endpoint: NSXPCListenerEndpoint) {
        let defaultTag = "tag"
        setEndpoint(tag: defaultTag, endpoint: endpoint)

        // 被标记的group, 才允许调用leave, 防止crash
        if getWaiterFlag(tag: defaultTag) {
            getWaiter(tag: defaultTag).leave()
            setWaiterFlag(tag: defaultTag, f: false)
            // 清理waiter
            setWaiter(tag: defaultTag, w: nil)
        } else {
            NSLog("Waiter flag is false, may be the endpoint query is time out.")
        }
    }
}

代码不复杂，其中tag是固定的字符串，queryEndpoint接口做了同步等待，等待服务进程提交匿名端口后返回给主程序。

合法性校验

XPC本身没有提供什么机制来判断通讯发起方是否合法，但是XPC提供了一个主动关闭连接的功能，所以可以借助这个功能，在服务进程做校验。当服务进程发现有连接过来时，可以通过判断请求方的进程信息，或者调用请求方事先约定的校验接口获取请求方相关校验信息，对于不合法的的请求方，服务进程可以随时关闭连接并永久绝再次连接。

下面是校验的时序图，包括服务进程对主程序的合法性校验以及主程序自行确认版本号是否匹配。

上面也是用到了Mac的全局通知，主程序订阅了全局通知，得到服务进程确认信息后才会继续往下走。

下面是服务进程匿名端口接收连接的Delegate代码。验证请求方合法性，并可以及时关闭非法请求。

class InternalAnonymousListenerDelegate: NSObject, NSXPCListenerDelegate {
    static let instance = InternalAnonymousListenerDelegate()

    func listener(_ listener: NSXPCListener, shouldAcceptNewConnection newConnection: NSXPCConnection) -> Bool {
        // This method is where the NSXPCListener configures, accepts, and resumes a new incoming NSXPCConnection.

        // Configure the connection.
        // First, set the interface that the exported object implements.
        newConnection.exportedInterface = NSXPCInterface(with: InternalAnonymousServiceProtocol.self)

        // Next, set the object that the connection exports. All messages sent on the connection to this service will be sent to the exported object to handle. The connection retains the exported object.
        newConnection.exportedObject = InternalAnonymousService.center

        newConnection.remoteObjectInterface = NSXPCInterface(with: MainAppXPCServiceProtocol.self)
        let appServ = newConnection.remoteObjectProxyWithErrorHandler { (err) in
            newConnection.invalidate()
        } as! MainAppXPCServiceProtocol //主程序暴露的协议
        InternalAnonymousXPC.instance.appServ = Serv
        // Resuming the connection allows the system to deliver more incoming messages.
        newConnection.resume()

        //这里也可以直接判断进程信息来确定进程是否合法 
        NSLog("ServiceDelegate: auditSessionIdentifier:\(newConnection.auditSessionIdentifier), processIdentifier:\(newConnection.processIdentifier)")

        if !InternalAnonymousService.acceptNewConnection {
            return false
        }

        // 对连接方进行校验
        sdkServ.reverseAuth { (key) in
            // 推送全局通知, 告知校验结果
            let nfcenter = CFNotificationCenterGetDistributedCenter()

            var authDic:[String:String]? = [:]
            if key != "key123" {
                InternalAnonymousService.acceptNewConnection = false
                newConnection.invalidate()
                authDic = nil
                InternalAnonymousXPC.instance.appServ = nil
                NSLog("xpc auth faild")
            }else {
                NSLog("xpc auth success")
            }

            CFNotificationCenterPostNotification(nfcenter, CFNotificationName("com.cocos.xpc.reverseAuth" as CFString), nil, authDic as CFDictionary?, false)
        }

        return InternalAnonymousService.acceptNewConnection
    }
}

上面演示了服务进程如何对主程序进行合法性校验。主程序可以调用服务进程提供的版本检查接口，确定主程序是否能和服务进程正常通讯。毕竟如果服务进程的版本不兼容，那么XPC通讯约定的Protocol就没法正常调用了。

多线程注意事项

涉及到的多线程编码主要是要防止死锁和错误调用DispatchGroup 。

上面合法性检验代码设计中使用了全局通知，有个需要注意的地方 ，订阅全局通知后，只能在主线程接收回调，为了避免死锁，主程序 向服务进程发起连接的所有代码都需要放到非主线程 执行。这个设计也是合理的，毕竟连接可能需要点时间。

另外一个地方是DispatchGroup ，使用DispatchGroup 做线程同步时，enter 和leave 两个调用必须是一对一，要特别注意别多调用leave ，也要注意及时释放DispatchGroup 。

waiter.enter() // DispatchGroup
var verAccept: Bool?
servXPC.submitVersion(ver: "1.0.0", withReply: { [weak waiter] accept in
    if accept {
        SDKLog("version accept!")
    } else {
        SDKLog("version not accept!")
    }
    verAccept = accept
    waiter?.leave()
})

if waiter.wait(timeout: .now() + .seconds(SDKTimeOut)) == .timedOut {
    SDKLog("version verify timeout.")
    return mainFailed(.failed)
}

比如上面submitVersion，是服务端XPC定义的接口，做线程同步的时候，先对group做一次leave ，但是回调必包被出发的时间是不确定的，主要取决XPC服务启动和响应的时间，所以需要用一个weak告知编译器弱引用waiter变量，并在waiter为nil时不调用leave 。

上文说的中介XPC中也有类似处理，这里就不再复述了。

总结

本文主要是提出了多进程通讯中的XPC技术的常见需求的解决方案，对于XPC具体代码的编写没有太多涉及，可以参考上文提供的文章，那里有详细的使用教程。

C的博客

不识庐山真面目,只缘身在此山中.

Mac上XPC多进程通讯的完整解决方案

Mac上XPC多进程通讯的完整解决方案

文档更新说明

前言

双向通讯

合法性校验

多线程注意事项

总结