HoneyBeeを使用したSwiftの高度な同時実行

公開: 2022-03-11

Swiftでの並行アルゴリズムの設計、テスト、および保守は困難であり、詳細を正しく取得することは、アプリの成功にとって非常に重要です。並行アルゴリズム（並列プログラミングとも呼ばれます）は、複数の（おそらく多くの）操作を同時に実行して、より多くのハードウェアリソースを活用し、全体的な実行時間を短縮するように設計されたアルゴリズムです。

Appleのプラットフォームでは、並行アルゴリズムを作成する従来の方法はNSOperationです。 NSOperationの設計により、プログラマーは並行アルゴリズムを個々の長時間実行される非同期タスクに細分化することができます。各タスクはNSOperationの独自のサブクラスで定義され、それらのクラスのインスタンスは、実行時にタスクの半順序を作成するために目的のAPIを介して結合されます。並行アルゴリズムを設計するこの方法は、7年間Appleのプラットフォームの最先端でした。

2014年、Appleは、並行操作の表現における劇的な前進として、Grand Central Dispatch（GCD）を導入しました。 GCDは、それに付随して機能する新しい言語機能ブロックとともに、非同期要求の開始直後に非同期応答ハンドラーをコンパクトに記述する方法を提供しました。プログラマーは、並行タスクの定義を多数のNSOperationサブクラスの複数のファイルに分散することを推奨されなくなりました。これで、並行アルゴリズム全体を1つのメソッド内で実行可能に記述できるようになりました。この表現力と型安全性の向上は、概念の大きな前進でした。この書き方の典型的なアルゴリズムは、次のようになります。

 func processImageData(completion: (result: Image?, error: Error?) -> Void) { loadWebResource("dataprofile.txt") { (dataResource, error) in guard let dataResource = dataResource else { completion(nil, error) return } loadWebResource("imagedata.dat") { (imageResource, error) in guard let imageResource = imageResource else { completion(nil, error) return } decodeImage(dataResource, imageResource) { (imageTmp, error) in guard let imageTmp = imageTmp else { completion(nil, error) return } dewarpAndCleanupImage(imageTmp) { imageResult in guard let imageResult = imageResult else { completion(nil, error) return } completion(imageResult, nil) } } } } }

このアルゴリズムを少し分解してみましょう。関数processImageDataは、作業を完了するために独自の4つの非同期呼び出しを行う非同期関数です。 4つの非同期呼び出しは、ブロックベースの非同期処理にとって最も自然な方法で、一方が他方の内部にネストされます。結果ブロックにはそれぞれオプションのErrorパラメーターがあり、1つを除くすべてに、aysnc操作の結果を示す追加のオプションパラメーターが含まれています。

上記のコードブロックの形は、おそらくほとんどのSwift開発者にはおなじみのようです。しかし、このアプローチの何が問題になっていますか？次の問題点のリストは、おそらく同じようによく知られています。

ネストされたコードブロックのこの「運命のピラミッド」形状は、すぐに扱いにくくなる可能性があります。さらに2つの非同期操作を追加するとどうなりますか？四？条件付き演算はどうですか？再試行動作またはリソース制限の保護はどうですか？実世界のコードは、ブログ投稿の例ほどクリーンでシンプルではありません。「破滅のピラミッド」効果により、コードが読みにくく、保守が難しく、バグが発生しやすくなります。
上記の例でのエラー処理の試みは、Swiftyですが、実際には不完全です。プログラマーは、2パラメーターのObjective-Cスタイルの非同期コールバックブロックが常に2つのパラメーターの1つを提供すると想定しています。両方が同時にゼロになることはありません。これは安全な仮定ではありません。並行アルゴリズムは、記述とデバッグが難しいことで有名であり、根拠のない仮定がその理由の一部です。完全で正しいエラー処理は、現実の世界で動作することを意図した並行アルゴリズムにとって避けられない必要性です。
この考えをさらに進めると、おそらく、呼び出された非同期関数を作成したプログラマーは、あなたほど原理的ではありませんでした。呼び出された関数がコールバックに失敗する条件がある場合はどうなりますか？または、複数回コールバックしますか？このような状況では、processImageDataの正確性はどうなりますか？プロはチャンスをつかみません。ミッションクリティカルな機能は、サードパーティによって作成された機能に依存している場合でも正しくなければなりません。
おそらく最も説得力のある、考慮されている非同期アルゴリズムは最適に構築されていません。最初の2つの非同期操作は、どちらもリモートリソースのダウンロードです。相互依存性はありませんが、上記のアルゴリズムはダウンロードを並行してではなく順次実行します。この理由は明らかです。ネストされたブロック構文は、そのような無駄を助長します。競争の激しい市場は、不必要な遅れを容認しません。アプリが非同期操作をできるだけ速く実行しない場合、別のアプリが実行します。

どうすればもっとうまくできるでしょうか？ HoneyBeeは、Swiftの並行プログラミングを簡単、表現力豊か、そして安全にするfutures/promisesライブラリです。上記の非同期アルゴリズムをHoneyBeeで書き直して、結果を調べてみましょう。

 func processImageData(completion: (result: Image?, error: Error?) -> Void) { HoneyBee.start() .setErrorHandler { completion(nil, $0) } .branch { stem in stem.chain(loadWebResource =<< "dataprofile.txt") + stem.chain(loadWebResource =<< "imagedata.dat") } .chain(decodeImage) .chain(dewarpAndCleanupImage) .chain { completion($0, nil) } }

この実装が開始する最初の行は、新しいHoneyBeeレシピです。 2行目は、デフォルトのエラーハンドラを確立します。 HoneyBeeレシピでは、エラー処理はオプションではありません。何かがうまくいかない場合は、アルゴリズムがそれを処理する必要があります。 3行目は、並列実行を可能にするブランチを開きます。 loadWebResourceの2つのチェーンが並行して実行され、それらの結果が結合されます（5行目）。ロードされた2つのリソースの結合された値は、完了が呼び出されるまでチェーンのdecodeImageなどに転送されます。

上記の問題点のリストを見ていき、HoneyBeeがこのコードをどのように改善したかを見てみましょう。この機能の保守が大幅に簡単になりました。 HoneyBeeレシピは、それが表現するアルゴリズムのように見えます。コードは読みやすく、理解しやすく、すばやく変更できます。 HoneyBeeの設計により、命令の順序を間違えると、実行時エラーではなく、コンパイル時エラーが発生します。この関数は、バグや人為的エラーの影響を受けにくくなりました。

発生する可能性のあるすべてのランタイムエラーは完全に処理されています。 HoneyBeeがサポートするすべての関数シグネチャ（38個あります）は、完全に処理されることが保証されています。この例では、Objective-Cスタイルの2パラメーターコールバックは、エラーハンドラーにルーティングされる非nilエラーを生成するか、チェーンを下に進む非nil値を生成するか、または両方の場合値はnilですHoneyBeeは、関数コールバックがそのコントラクトを実行していないことを説明するエラーを生成します。

HoneyBeeは、関数のコールバックが呼び出される回数の契約上の正確さも処理します。関数がコールバックの呼び出しに失敗した場合、HoneyBeeは記述的な失敗を生成します。関数がコールバックを複数回呼び出す場合、HoneyBeeは補助的な呼び出しを抑制して警告をログに記録します。これらの障害応答（およびその他）は両方とも、プログラマーの個々のニーズに合わせてカスタマイズできます。

うまくいけば、この形式のprocessImageDataがリソースのダウンロードを適切に並列化して、最適なパフォーマンスを提供することはすでに明らかです。 HoneyBeeの最も強力な設計目標の1つは、レシピがそれが表現するアルゴリズムのように見えることです。

ずっといい。右？しかし、HoneyBeeにはさらに多くの機能があります。

警告：次のケーススタディは、気の弱い人向けではありません。次の問題の説明を検討してください。モバイルアプリはCoreDataを使用して状態を維持します。メディアと呼ばれるNSManagedObjectモデルがあります。これは、バックエンドサーバーにアップロードされたメディアアセットを表します。ユーザーは、一度に数十のメディアアイテムを選択し、それらをバッチでバックエンドシステムにアップロードすることができます。メディアは最初に参照文字列を介して表されます。参照文字列はMediaオブジェクトに変換する必要があります。幸い、アプリにはすでにそれを実行するヘルパーメソッドが含まれています。

 func export(_ mediaRef: String, completion: @escaping (Media?, Error?) -> Void) { // transcoding stuff completion(Media(context: managedObjectContext), nil) }

メディア参照がMediaオブジェクトに変換されたら、メディアアイテムをバックエンドにアップロードする必要があります。ここでも、ネットワーク関連の処理を実行するためのヘルパー関数が用意されています。

 func upload(_ media: Media, completion: @escaping (Error?) -> Void) { // network stuff completion(nil) }

ユーザーは一度に数十のメディアアイテムを選択できるため、UXデザイナーはアップロードの進行状況についてかなり強力なフィードバックを指定しています。要件は、次の4つの機能にまとめられています。

 /// Called if anything goes wrong in the upload func errorHandler(_ error: Error) { // do the right thing } /// Called once per mediaRef, after either a successful or unsuccessful upload func singleUploadCompletion(_ mediaRef: String) { // update a progress indicator } /// Called once per successful upload func singleUploadSuccess(_ media: Media) { // do celebratory things } /// Called if the entire batch was considered to be uploaded successfully. func totalProcessSuccess() { // declare victory }

ただし、アプリは期限切れになることがあるメディア参照をソースするため、アップロードの少なくとも半分が成功した場合、ビジネスマネージャーはユーザーに「成功」メッセージを送信することを決定しました。つまり、試行されたアップロードの半分未満が失敗した場合、並行プロセスは勝利を宣言し、 totalProcessSuccessを呼び出す必要があります。これは、開発者として渡された仕様です。しかし、経験豊富なプログラマーとして、適用しなければならない要件がもっとあることに気づきます。

もちろん、Businessはバッチアップロードをできるだけ早く実行することを望んでいるため、シリアルアップロードは問題外です。アップロードは並行して実行する必要があります。

しかしあまりありません。バッチ全体を無差別にasync化すると、数十の同時アップロードがモバイルNIC（ネットワークインターフェイスカード）にフラッディングし、アップロードは実際にはシリアルよりも遅くなり、速くはなりません。

モバイルネットワーク接続は安定しているとは見なされません。短いトランザクションでも、ネットワーク接続の変更のみが原因で失敗する可能性があります。アップロードが失敗したことを本当に宣言するには、少なくとも1回はアップロードを再試行する必要があります。

再試行ポリシーには、一時的な障害が発生しないため、エクスポート操作を含めないでください。

エクスポートプロセスはコンピューティングバウンドであるため、メインスレッドから実行する必要があります。

エクスポートはコンピューティングバウンドであるため、プロセッサのスラッシングを回避するために、残りのアップロードプロセスよりも同時インスタンスの数を少なくする必要があります。

上記の4つのコールバック関数はすべてUIを更新するため、すべてメインスレッドで呼び出す必要があります。

メディアはNSManagedObjectであり、 NSManagedObjectContextに由来し、尊重する必要のある独自のスレッド要件があります。

この問題の仕様は少しあいまいに見えますか？このような問題が将来潜んでいることに気付いても驚かないでください。私は自分の仕事でこのようなものに出会いました。まず、従来のツールを使用してこの問題を解決してみましょう。バックルアップ、これはきれいではありません。

 /// An enum describing specific problems that the algorithm might encounter. enum UploadingError : Error { case invalidResponse case tooManyFailures } /// A semaphore to prevent flooding the NIC let outerLimit = DispatchSemaphore(value: 4) /// A semaphore to prevent thrashing the processor let exportLimit = DispatchSemaphore(value: 1) /// The number of times to retry the upload if it fails let uploadRetries = 1 /// Dispatch group to keep track of when the entire process is finished let fullProcessDispatchGroup = DispatchGroup() /// How many of the uploads fully completed. var uploadSuccesses = 0 // this notify block is called when the full process has completed. fullProcessDispatchGroup.notify(queue: DispatchQueue.main) { let successRate = Float(uploadSuccesses) / Float(mediaReferences.count) if successRate > 0.5 { totalProcessSuccess() } else { errorHandler(UploadingError.tooManyFailures) } } // start in the background DispatchQueue.global().async { for mediaRef in mediaReferences { // alert the group that we're starting a process fullProcessDispatchGroup.enter() // wait until it's safe to start uploading outerLimit.wait() /// common cleanup operations needed later func finalizeMediaRef() { singleUploadCompletion(mediaRef) fullProcessDispatchGroup.leave() outerLimit.signal() } // wait until it's safe to start exporting exportLimit.wait() export(mediaRef) { (media, error) in // allow another export to begin exportLimit.signal() if let error = error { DispatchQueue.main.async { errorHandler(error) finalizeMediaRef() } } else { guard let media = media else { DispatchQueue.main.async { errorHandler(UploadingError.invalidResponse) finalizeMediaRef() } return } // the export was successful var uploadAttempts = 0 /// define the upload process and its retry behavior func doUpload() { // respect Media's threading requirements managedObjectContext.perform { upload(media) { error in if let error = error { if uploadAttempts < uploadRetries { uploadAttempts += 1 doUpload() // retry } else { DispatchQueue.main.async { // too many upload failures errorHandler(error) finalizeMediaRef() } } } else { DispatchQueue.main.async { uploadSuccesses += 1 singleUploadSuccess(media) finalizeMediaRef() } } } } } // kick off the first upload doUpload() } } } }

うわー！コメントなしで、それは約75行です。ずっと推論に従いましたか？新しい仕事で最初の週にこのモンスターに遭遇した場合、どのように感じますか？それを維持する準備ができていると思いますか、それとも変更しますか？エラーが含まれているかどうかわかりますか？エラーが含まれていますか？

ここで、HoneyBeeの代替案を検討してください。

 HoneyBee.start(on: DispatchQueue.main) .setErrorHandler(errorHandler) .insert(mediaReferences) .setBlockPerformer(DispatchQueue.global()) .each(limit: 4, acceptableFailure: .ratio(0.5)) { elem in elem.finally { link in link.setBlockPerformer(DispatchQueue.main) .chain(singleUploadCompletion) } .limit(1) { link in link.chain(export) } .setBlockPerformer(managedObjectContext) .retry(1) { link in link.chain(upload) // subject to transient failure } .setBlockPerformer(DispatchQueue.main) .chain(singleUploadSuccess) } .setBlockPerformer(DispatchQueue.main) .drop() .chain(totalProcessSuccess)

このフォームはどのようにあなたを襲いますか？少しずつ見ていきましょう。最初の行で、メインスレッドから始めてHoneyBeeレシピを開始します。メインスレッドから開始することで、すべてのエラーがメインスレッドのerrorHandler（2行目）に渡されるようにします。 3行目は、 mediaReferences配列をプロセスチェーンに挿入します。次に、並列処理に備えてグローバルバックグラウンドキューに切り替えます。 5行目で、各mediaReferencesに対して並列反復を開始します。この並列処理は、最大4つの同時操作に制限されます。また、サブチェーンの少なくとも半分が成功した場合（エラーを起こさないでください）、完全な反復が成功したと見なされることを宣言します。 6行目は、以下のサブチェーンが成功するか失敗するかに関係なく呼び出されるfinallyリンクを宣言しています。 finallyのリンクで、メインスレッドに切り替え（7行目）、 singleUploadCompletionを呼び出します（8行目）。 10行目では、エクスポート操作（11行目）の周囲に最大並列化を1（単一実行）に設定しています。 13行目は、 managedObjectContextインスタンスが所有するプライベートキューに切り替わります。 14行目は、アップロード操作の1回の再試行を宣言しています（15行目）。 17行目は再びメインスレッドに切り替わり、18行目はsingleUploadSuccessを呼び出します。ライン20が実行されるまでに、すべての並列反復が完了します。失敗した反復の半分未満の場合、行20は最後にもう一度メインキューに切り替わり（それぞれがバックグラウンドキューで実行されたことを思い出してください）、21はインバウンド値を削除し（まだmediaReferences ）、22はtotalProcessSuccessを呼び出します。

HoneyBeeフォームは、保守が簡単なことは言うまでもなく、より明確で、よりクリーンで、読みやすくなっています。 Mediaオブジェクトをマップ関数のような配列に再統合するためにループが必要な場合、このアルゴリズムの長い形式はどうなりますか？変更を加えた後、アルゴリズムのすべての要件がまだ満たされていることをどの程度確信できますか？ HoneyBeeフォームでは、この変更は、並列マップ関数を使用するためにそれぞれをマップに置き換えることです。（はい、それも減少しています。）

HoneyBeeは、Swiftの強力な先物ライブラリであり、非同期および並行アルゴリズムの記述をより簡単、安全、表現力豊かにします。この記事では、HoneyBeeを使用して、アルゴリズムの保守をより簡単に、より正確に、より高速にする方法について説明しました。 HoneyBeeは、再試行サポート、複数のエラーハンドラー、リソースガード、コレクション処理（非同期形式のmap、filter、reduce）などの他の主要な非同期パラダイムもサポートしています。機能の完全なリストについては、Webサイトを参照してください。詳細や質問については、新しいコミュニティフォーラムをご覧ください。

付録：非同期関数の契約上の正確性の確保

機能の契約上の正確さを保証することは、コンピュータサイエンスの基本的な信条です。事実上すべての最新のコンパイラーが、値を返すことを宣言する関数が1回だけ戻ることを確認するためのチェックを持っているほどです。 1回未満または複数回の戻りはエラーとして扱われ、完全なコンパイルを適切に防ぎます。

ただし、このコンパイラ支援は通常、非同期関数には適用されません。次の（遊び心のある）例を考えてみましょう。

 func generateIcecream(from int: Int, completion: (String) -> Void) { if int > 5 { if int < 20 { completion("Chocolate") } else if int < 10 { completion("Strawberry") } completion("Pistachio") } else if int < 2 { completion("Vanilla") } }

generateIcecream関数はIntを受け入れ、非同期で文字列を返します。いくつかの明らかな問題が含まれている場合でも、swiftコンパイラは上記の形式を正しいものとして受け入れます。特定の入力が与えられると、この関数は完了を0回、1回、または2回呼び出す場合があります。非同期関数を使用したことのあるプログラマーは、自分の作業でこの問題の例を思い出すことがよくあります。私たちは何ができる？確かに、コードをリファクタリングしてきれいにすることができます（範囲ケースのあるスイッチはここで機能します）。ただし、機能の複雑さを軽減するのが難しい場合があります。定期的に関数を返す場合と同じように、コンパイラーが正確さの検証を支援してくれるとよいのではないでしょうか。

方法があることがわかりました。次の迅速な呪文を守ってください。

 func generateIcecream(from int: Int, completion: (String) -> Void) { let finalResult: String defer { completion(finalResult) } let completion: Void = Void() defer { completion } if int > 5 { if int < 20 { completion("Chocolate") } else if int < 10 { completion("Strawberry") } // else completion("Pistachio") } else if int < 2 { completion("Vanilla") } }

この関数の先頭に挿入された4行は、完了コールバックが1回だけ呼び出されたことをコンパイラーに確認させます。これは、この関数がコンパイルされなくなったことを意味します。どうしたの？最初の行では、この関数で最終的に生成する結果を宣言しますが、初期化しません。未定義のままにしておくと、使用する前に1回割り当てる必要があり、宣言することで、2回割り当てられないようにすることができます。 2行目は、この関数の最後のアクションとして実行される延期です。関数の残りの部分によって割り当てられた後、 finalResultを使用して完了ブロックを呼び出します。 3行目は、コールバックパラメータをシャドウするcompletionという新しい定数を作成します。新しい補完は、パブリックAPIを宣言しないVoidタイプです。この行は、この行の後に補完を使用するとコンパイラエラーになることを保証します。 2行目の延期は、完了ブロックの唯一の許可された使用法です。 4行目では、新しい完了定数が使用されていない場合に発生するコンパイラ警告を削除しています。

そのため、swiftコンパイラに、この非同期関数がそのコントラクトを実行していないことを報告するように強制することに成功しました。それを正しくするための手順を見ていきましょう。まず、コールバックへのすべての直接アクセスをfinalResultへの割り当てに置き換えましょう。

 func generateIcecream(from int: Int, completion: (String) -> Void) { let finalResult: String defer { completion(finalResult) } let completion: Void = Void() defer { completion } if int > 5 { if int < 20 { finalResult = "Chocolate" } else if int < 10 { finalResult = "Strawberry" } // else finalResult = "Pistachio" } else if int < 2 { finalResult = "Vanilla" } }

現在、コンパイラは2つの問題を報告しています。

 error: AsyncCorrectness.playground:1:8: error: constant 'finalResult' used before being initialized defer { completion(finalResult) } ^ error: AsyncCorrectness.playground:11:3: error: immutable value 'finalResult' may only be initialized once finalResult = "Pistachio"

予想どおり、この関数には、 finalResultが0回割り当てられるパスウェイと、複数回割り当てられるパスウェイがあります。これらの問題は次のように解決します。

 func generateIcecream(from int: Int, completion: (String) -> Void) { let finalResult: String defer { completion(finalResult) } let completion: Void = Void() defer { completion } if int > 5 { if int < 20 { finalResult = "Chocolate" } else if int < 10 { finalResult = "Strawberry" } else { finalResult = "Pistachio" } } else if int < 2 { finalResult = "Vanilla" } else { finalResult = "Neapolitan" } }

「ピスタチオ」は適切なelse節に移動され、一般的なケース（もちろん「ネアポリタン」）をカバーできなかったことに気づきました。

今説明したパターンは、オプションの値、オプションのエラー、または一般的な結果列挙型のような複合型を返すように簡単に調整できます。コールバックが1回だけ呼び出されることを確認するようにコンパイラーを強制することにより、非同期関数の正確性と完全性を主張できます。