Coroutinesを中心としたAndroidアプリでの並行数制限・排他制御

2024/12/19 pixiv App Night発表資料

Transcript

1. 2024-12-19 pixiv App Night / fusuma CoroutinesΛத৺ͱͨ͠AndroidΞϓϦͰͷ ฒߦ਺੍ݶɾഉଞ੍ޚ
2. ࣗݾ঺հ: fusuma(;͢·) • ϐΫγϒגࣜձࣾ • 2021೥7݄த్ೖࣾ • ίϛοΫࣄۀ෦ AndroidΤϯδχΞ
3. CoroutineʹΑΔฒߦॲཧͷαϙʔτ • ฒߦॲཧ͸೉͍͠ • ࣮ߦॱͷ༧ଌͷࠔ೉͞ɺಉ࣌ʹॲཧͯ͠໰୊ͳ͍͔……ͳͲ • kotlinx.coroutines͸ฒߦॲཧΛαϙʔτ͢Δ࣮૷͕๛෋ • ࠓ೔ͷτϐοΫ͸ฒߦॲཧ਺ͷ੍ݶɾഉଞ੍ޚ
4. ࠓ೔ͷτϐοΫ 1. SemaphoreʹΑΔฒߦॲཧ਺ͷ੍ݶ 2. MutexʹΑΔഉଞ੍ޚ ͓୊: ෳ਺ͷϑΝΠϧΛฒߦͰμ΢ϯϩʔυ͢ΔϑΝΠϧμ΢ϯϩʔμʔͷ࣮૷
5. ϑΝΠϧμ΢ϯϩʔμʔͷ࢓༷ 1. 3ϑΝΠϧ·ͰฒߦͰμ΢ϯϩʔυ͢Δ 2. μ΢ϯϩʔυதʹϑΝΠϧͷμ΢ϯϩʔυॱΛมߋͰ͖Δ(ࢦఆͨ͠ ϑΝΠϧΛ༏ઌμ΢ϯϩʔυ)
6. ϑΝΠϧμ΢ϯϩʔμʔͷ࢓༷ 1. 3ϑΝΠϧ·ͰฒߦͰμ΢ϯϩʔυ͢Δ 2. μ΢ϯϩʔυதʹϑΝΠϧͷμ΢ϯϩʔυॱΛมߋͰ͖Δ(ࢦఆͨ͠ ϑΝΠϧΛ༏ઌͯ͠μ΢ϯϩʔυ)
7. ϑΝΠϧμ΢ϯϩʔμʔ class FileDownloader(private val urlList: List<URL>) { suspend fun downloadFiles() = withContext(Dispatchers.IO) { urlList.forEach { download(it) } } } suspend fun download(url: URL) { // ダウンロード処理 }
8. ฒߦͰμ΢ϯϩʔυ class FileDownloader(private val urlList: List<URL>) { suspend fun downloadFiles() = withContext(Dispatchers.IO) { urlList.forEach { launch { download(it) } } } } ʮ3ϑΝΠϧ·ͰʯฒߦͰμ΢ϯϩʔυͱ͍͏࢓༷Λຬ͍ͨͯ͠ͳ͍
9. 1. SemaphoreʹΑΔฒߦॲཧ਺ͷ੍ݶ
10. Semaphore • Semaphore(permits: Int)ͰڐՄ͢Δॲཧͷ਺Λࢦఆ͢Δ • ओͳϝιου • acquire() • ڐՄΛ1ͭऔಘɻऔಘͰ͖ͳ͍৔߹ɺڐՄ͕ղ์͞ΕΔ·Ͱ଴ػ • suspendؔ਺ʹͳ͓ͬͯΓ଴ػ࣌ʹεϨουΛϒϩοΫ͠ͳ͍ • release() • ڐՄΛ1ͭղ์͢Δ • withPermit() • acquireΛݺͼग़͠ɺ౉ͨ͠lambdaΛ࣮ߦ͢ΔͱࣗಈͰrelease(ྫ֎ൃੜ࣌΋release͞ΕΔ)
11. SemaphoreͰฒߦॲཧ਺Λ੍ݶ class FileDownloader(private val urlList: List<URL>) { val maxDownloadsSemaphore = Semaphore(permits = 3) suspend fun downloadFiles() = withContext(Dispatchers.IO) { urlList.forEach { launch { maxDownloadsSemaphore.withPermit { download(it) } } } } } SemaphoreͰڐՄ͢Δॲཧͷ਺Λࢦఆ withPermitͰμ΢ϯϩʔυॲཧ ڐՄ͞Εͨॲཧ਺Λ௒͑Δ৔߹͸ɺ ॲཧ਺ͷۭ͖͕Ͱ͖Δ·Ͱதஅ͞ΕΔ
12. ϑΝΠϧμ΢ϯϩʔμʔͷ࢓༷ 1. 3ϑΝΠϧ·ͰฒߦͰμ΢ϯϩʔυ͢Δ 2. μ΢ϯϩʔυதʹϑΝΠϧͷμ΢ϯϩʔυॱΛมߋͰ͖Δ(ࢦఆͨ͠ ϑΝΠϧΛ༏ઌμ΢ϯϩʔυ)
13. ࢦఆͨ͠ཁૉΛઌ಄ʹҠಈ͢ΔQueue class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() // 先頭のURLを取り出す(ダウンローダーはpollを呼び出して順次ダウンロードする) fun poll(): URL? = urlList.removeFirstOrNull() // 指定したURLを先頭に移動する(優先ダウンロード) fun moveToHead(url: URL) { if (urlList.remove(url)) { urlList.add(0, url) } } }
14. ෳ਺ͷcoroutine͔ΒQueueΛૢ࡞ class FileDownloader(private val urlList: List<URL>) { val maxDownloadsSemaphore = Semaphore(permits = 3) val urlQueue = UrlQueue(urlList) suspend fun downloadFiles() = withContext(Dispatchers.IO) { repeat(urlList.size) { launch { maxDownloadsSemaphore.withPermit { val url = urlQueue.poll() ?: return@launch download(url) } } } } fun prioritizeUrl(url: URL) { urlQueue.moveToHead(url) } } class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() fun poll(): URL? = urlList.removeFirstOrNull() fun moveToHead(url: URL) { if (urlList.remove(url)) { urlList.add(0, url) } } }
15. ෳ਺ͷcoroutine͔ΒQueueΛૢ࡞ class FileDownloader(private val urlList: List<URL>) { val maxDownloadsSemaphore = Semaphore(permits = 3) val urlQueue = UrlQueue(urlList) suspend fun downloadFiles() = withContext(Dispatchers.IO) { repeat(urlList.size) { launch { maxDownloadsSemaphore.withPermit { val url = urlQueue.poll() ?: return@launch download(url) } } } } fun prioritizeUrl(url: URL) { urlQueue.moveToHead(url) } } class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() fun poll(): URL? = urlList.removeFirstOrNull() fun moveToHead(url: URL) { if (urlList.remove(url)) { urlList.add(0, url) } } } Queueͷૢ࡞͕εϨουηʔϑͰ͸ͳ͍
16. σʔλڝ߹ͱڝ߹ঢ়ଶ • σʔλڝ߹(Data Race) • ಉ࣌ʹڞ༗σʔλʹΞΫηε͠ɺগͳ͘ͱ΋1͕ͭॻ͖ࠐΈΛߦ͏৔߹ʹൃ ੜ͢Δ • ى͖Δ͜ͱ: ༧ଌෆೳͳ໰୊͕ى͜Δ • ڝ߹ঢ়ଶ(Race Condition) • ࣮ߦॱং΍λΠϛϯάʹґଘͨ͠ҙਤ͠ͳ͍ग़ྗͷมԽ • ى͖Δ͜ͱɿಉ͡ೖྗͰ΋ҟͳΔ݁ՌΛग़ྗ͢Δɾ݁Ռͷ੔߹ੑ͕औΕͳ͍
17. εϨουηʔϑͰͳ͍ॲཧ class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() fun poll(): URL? = urlList.removeFirstOrNull() … } Coroutine A の動作 Coroutine B の動作 urlList の状態 1 poll() を呼び出し先頭を読む(URL1) poll() を呼び出し先頭を読む (URL1) [URL1, URL2, URL3] 2 先頭を削除(URL1) [URL2, URL3] 3 URL1を返す 先頭を削除(URL2) [URL3] 4 URL1を返す
18. εϨουηʔϑͰͳ͍ॲཧ class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() fun poll(): URL? = urlList.removeFirstOrNull() … } Coroutine A の動作 Coroutine B の動作 urlList の状態 1 poll() を呼び出し先頭を読む(URL1) poll() を呼び出し先頭を読む (URL1) [URL1, URL2, URL3] 2 先頭を削除(URL1) [URL2, URL3] 3 URL1を返す 先頭を削除(URL2) [URL3] 4 URL1を返す
19. εϨουηʔϑͰͳ͍ॲཧ class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() fun poll(): URL? = urlList.removeFirstOrNull() … } Coroutine A の動作 Coroutine B の動作 urlList の状態 1 poll() を呼び出し先頭を読む(URL1) poll() を呼び出し先頭を読む (URL1) [URL1, URL2, URL3] 2 先頭を削除(URL1) [URL2, URL3] 3 URL1を返す 先頭を削除(URL2) [URL3] 4 URL1を返す
20. ڝ߹ঢ়ଶ͕ൃੜ͠ɺظ଴ͱҟͳΔ݁Ռʹ class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() fun poll(): URL? = urlList.removeFirstOrNull() … } Coroutine A の動作 Coroutine B の動作 urlList の状態 1 poll() を呼び出し先頭を読む(URL1) poll() を呼び出し先頭を読む (URL1) [URL1, URL2, URL3] 2 先頭を削除(URL1) [URL2, URL3] 3 URL1を返す 先頭を削除(URL2) [URL3] 4 URL1を返す
21. 2. MutexʹΑΔഉଞ੍ޚ
22. Mutex • ϩοΫΛऔಘ͠ɺಛఆͷίʔυ͕ಉ࣌ʹ࣮ߦ͞ΕΔ͜ͱΛ๷͙ • ओͳϝιου • lock() • ϩοΫΛऔΔɻϩοΫ͕औΕͳ͍৔߹ɺϩοΫղআ͞ΕΔ·Ͱ଴ػ • suspendؔ਺ʹͳ͓ͬͯΓɺ଴ػ࣌ʹεϨουΛϒϩοΫ͠ͳ͍ • unlock() • ϩοΫΛղআ • withLock() • lockΛݺͼग़͠ɺ౉ͨ͠lambdaΛ࣮ߦ͢ΔͱࣗಈͰunlock(ྫ֎ൃੜ࣌΋unlock)
23. MutexͰഉଞ੍ޚ class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() private val safeAccessMutex = Mutex() suspend fun poll(): URL? = safeAccessMutex.withLock { urlList.removeFirstOrNull() } suspend fun moveToHead(url: URL) = safeAccessMutex.withLock { if (urlList.remove(url)) { urlList.add(0, url) } } }
24. MutexͰഉଞ੍ޚ class UrlQueue(urlList: List<URL>) { private val urlList = urlList.toMutableList() private val safeAccessMutex = Mutex() suspend fun poll(): URL? = safeAccessMutex.withLock { urlList.removeFirstOrNull() } suspend fun moveToHead(url: URL) = safeAccessMutex.withLock { if (urlList.remove(url)) { urlList.add(0, url) } } } MutexͰϩοΫΛऔΓಉ࣌ʹϦιʔεʹ ΞΫηε͞Εͳ͍Α͏ʹ͢Δ(ഉଞ੍ޚ)
25. MutexͱSemaphoreͷؔ܎ • Mutex͸permits=1ͷSemaphore(ಉ࣌ʹॲཧͰ͖Δ਺͕1ͭͷΈ)ͱߟ͑ ΒΕΔ • ࣮ࡍਖ਼ʹͦͷΑ͏ʹ࣮૷͞Ε͍ͯΔ
26. ผղ: CoroutineScope.actor • 1ͭͷcoroutineΛىಈ͠ɺ಺෦ʹঢ়ଶͱॲཧΛΧϓηϧԽ͢Δ • ঢ়ଶͷಡΈॻ͖͕௚ྻͰ࣮ߦ͞ΕΔͷͰεϨουηʔϑ • actor͸SendChannelΛฦ͢ • ͜ͷchannelʹϝοηʔδΛૹ৴͢Δ͜ͱͰactorͷঢ়ଶߋ৽ɾॲཧ͕ ࣮ߦ͞ΕΔ
27. ผղ: CoroutineScope.actor sealed class Message { data class Poll(val result: CompletableDeferred<URL?>) : Message() data class MoveToHead(val url: URL) : Message() } @OptIn(ObsoleteCoroutinesApi::class) fun CoroutineScope.urlQueueActor(urlList: List<URL>) = actor<Message> { val urlList = urlList.toMutableList() for (msg in channel) { when (msg) { is Message.Poll -> msg.result.complete(urlList.removeFirstOrNull()) is Message.MoveToHead -> { if (urlList.remove(msg.url)) { urlList.add(0, msg.url) } } } } }
28. ผղ: CoroutineScope.actor sealed class Message { data class Poll(val result: CompletableDeferred<URL?>) : Message() data class MoveToHead(val url: URL) : Message() } @OptIn(ObsoleteCoroutinesApi::class) fun CoroutineScope.urlQueueActor(urlList: List<URL>) = actor<Message> { val urlList = urlList.toMutableList() for (msg in channel) { when (msg) { is Message.Poll -> msg.result.complete(urlList.removeFirstOrNull()) is Message.MoveToHead -> { if (urlList.remove(msg.url)) { urlList.add(0, msg.url) } } } } } actorͰड͚औΔϝοηʔδΛఆٛ
29. ผղ: CoroutineScope.actor sealed class Message { data class Poll(val result: CompletableDeferred<URL?>) : Message() data class MoveToHead(val url: URL) : Message() } @OptIn(ObsoleteCoroutinesApi::class) fun CoroutineScope.urlQueueActor(urlList: List<URL>) = actor<Message> { val urlList = urlList.toMutableList() for (msg in channel) { when (msg) { is Message.Poll -> msg.result.complete(urlList.removeFirstOrNull()) is Message.MoveToHead -> { if (urlList.remove(msg.url)) { urlList.add(0, msg.url) } } } } } ϝοηʔδΛड৴ͯ͠ॲཧ ௚ྻͰॲཧ͞ΕΔͷͰεϨουηʔϑ
30. ผղ: CoroutineScope.actor class UrlQueueActor(scope: CoroutineScope, urlList: List<URL>) { private val queueActor = scope.urlQueueActor(urlList) suspend fun poll(): URL? { val result = CompletableDeferred<URL?>() queueActor.send(Message.Poll(result)) return result.await() } suspend fun moveToHead(url: URL) { queueActor.send(Message.MoveToHead(url)) } } actorͷ໭Γ஋ͷChannelʹ ϝοηʔδΛૹ৴
31. CoroutineScope.actor͸ઃܭͷมߋ༧ఆ͋Γ • ͨͩ͠CoroutineScope.actor͸@ObsoleteCoroutinesApi • ࠶ઃܭ͕ߟ͑ΒΕ͍ͯΔ • ݴٴ͞Ε͍ͯΔissueʹ͸ಈ͖͕ͳ͔ͬͨΓ͢Δ(https://github.com/ Kotlin/kotlinx.coroutines/issues/87)
32. ·ͱΊ • kotlinx.coroutines͸ฒߦॲཧΛαϙʔτ͢Δ࣮૷͕ଟ͘ఏڙ͞Ε͍ͯΔ • SemaphoreͰฒߦॲཧ਺ͷ੍ݶ • MutexͰഉଞ੍ޚ • ͱ͸͍͑ฒߦॲཧ͸೉͍͠΋ͷ • ஫ҙਂ͍ίʔυϨϏϡʔͳͲ΋ॏཁ
33. Appendix
34. ϑΝΠϧμ΢ϯϩʔμʔͷ࢓༷ 1. 3ϑΝΠϧ·ͰฒߦͰμ΢ϯϩʔυ͢Δ 2. Ϣʔβʔ͕ࢦఆͨ͠ϑΝΠϧΛ༏ઌͯ͠μ΢ϯϩʔυ͢Δ 3. μ΢ϯϩʔυ͕׬ྃͨ͠ϑΝΠϧͷ਺Λ௨஌͢Δ(Viewʹ൓ө)
35. 3. StateFlowʹΑΔεϨουηʔϑͳঢ়ଶߋ৽
36. StateFlowͰμ΢ϯϩʔυ׬ྃ਺Λ௨஌ private val _downloadedCount = MutableStateFlow(0) val downloadedCount: StateFlow<Int> = _downloadedCount suspend fun downloadFiles() = withContext(Dispatchers.IO) { repeat(urlList.size) { launch { maxDownloadsSemaphore.withPermit { val url = urlQueue.poll() ?: return@launch download(url) _downloadedCount.value = _downloadedCount.value + 1 } } } } StateFlow.value͸ಡΈࠐΈɾॻ͖ࠐΈ͕ͦΕͧΕεϨουηʔϑʹͳ͍ͬͯΔͷͰɺ σʔλڝ߹͸ى͜Βͳ͍ɻ͔͠͠ಡΈࠐΈɾॻ͖ࠐΈ͕ΞτϛοΫͳૢ࡞Ͱ͸ͳ͍ͨΊɺ ࣮ߦॱʹΑͬͯ͸ڝ߹ঢ়ଶͱͳΔ
37. StateFlowʹΑΔεϨουηʔϑͳঢ়ଶߋ৽ private val _downloadedCount = MutableStateFlow(0) val downloadedCount: StateFlow<Int> = _downloadedCount suspend fun downloadFiles() = withContext(Dispatchers.IO) { repeat(urlList.size) { launch { maxDownloadsSemaphore.withPermit { val url = urlQueue.poll() ?: return@launch download(url) _downloadedCount.update { it + 1 } } } } } StateFlow.update͸ಡΈࠐΈɾॻ͖ࠐΈ͕ ΞτϛοΫʹͳ͓ͬͯΓɺڝ߹ঢ়ଶΛճආͰ͖Δ
38. ׬ྃ਺Λ௨஌͢ΔFlowΛฦ͢ύλʔϯ fun downloadFiles() = channelFlow { val downloadedCount = AtomicInteger(0) repeat(urlList.size) { launch { maxDownloadsSemaphore.withPermit { val url = urlQueue.poll() ?: return@launch download(url) send(downloadedCount.incrementAndGet()) } } } }.flowOn(Dispatchers.IO)
39. ׬ྃ਺Λ௨஌͢ΔFlowΛฦ͢ύλʔϯ import kotlinx.atomicfu.atomic … fun downloadFiles() = channelFlow { val downloadedCount = atomic(0) repeat(urlList.size) { launch { maxDownloadsSemaphore.withPermit { val url = urlQueue.poll() ?: return@launch download(url) send(downloadedCount.incrementAndGet()) } } } }.flowOn(Dispatchers.IO) kotlinx.atomicfu.AtomicIntͰ εϨουηʔϑͳߋ৽