feat(android): bound MetadataProvider on-miss fetches to 4 concurrent
This commit is contained in:
@@ -15,10 +15,14 @@ import kotlinx.coroutines.flow.Flow
|
|||||||
import kotlinx.coroutines.flow.map
|
import kotlinx.coroutines.flow.map
|
||||||
import kotlinx.coroutines.flow.onEach
|
import kotlinx.coroutines.flow.onEach
|
||||||
import kotlinx.coroutines.launch
|
import kotlinx.coroutines.launch
|
||||||
|
import kotlinx.coroutines.sync.Semaphore
|
||||||
|
import kotlinx.coroutines.sync.withPermit
|
||||||
import java.util.concurrent.ConcurrentHashMap
|
import java.util.concurrent.ConcurrentHashMap
|
||||||
import javax.inject.Inject
|
import javax.inject.Inject
|
||||||
import javax.inject.Singleton
|
import javax.inject.Singleton
|
||||||
|
|
||||||
|
private const val MAX_CONCURRENT_FETCHES = 4
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Cache-first metadata reads with eager on-miss live-fetch.
|
* Cache-first metadata reads with eager on-miss live-fetch.
|
||||||
*
|
*
|
||||||
@@ -49,6 +53,12 @@ class MetadataProvider @Inject constructor(
|
|||||||
private val inFlightArtists = ConcurrentHashMap<String, Job>()
|
private val inFlightArtists = ConcurrentHashMap<String, Job>()
|
||||||
private val inFlightTracks = ConcurrentHashMap<String, Job>()
|
private val inFlightTracks = ConcurrentHashMap<String, Job>()
|
||||||
|
|
||||||
|
// Caps concurrent on-miss fetches so a cold Home (~30 tiles) doesn't
|
||||||
|
// fire ~30 parallel /api calls. Mirrors Flutter HydrationQueue's
|
||||||
|
// _maxConcurrent = 4. Per-id dedup (above) is orthogonal: the gate
|
||||||
|
// bounds distinct-id fetches; extras suspend on the semaphore.
|
||||||
|
private val fetchGate = Semaphore(MAX_CONCURRENT_FETCHES)
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Live AlbumRef? for [id]. Fires a background fetch the first
|
* Live AlbumRef? for [id]. Fires a background fetch the first
|
||||||
* time the cache emits null. Safe to subscribe concurrently from
|
* time the cache emits null. Safe to subscribe concurrently from
|
||||||
@@ -98,7 +108,7 @@ class MetadataProvider @Inject constructor(
|
|||||||
): Job = inFlight.computeIfAbsent(id) {
|
): Job = inFlight.computeIfAbsent(id) {
|
||||||
appScope.launch {
|
appScope.launch {
|
||||||
try {
|
try {
|
||||||
block()
|
fetchGate.withPermit { block() }
|
||||||
} catch (
|
} catch (
|
||||||
@Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
|
@Suppress("TooGenericExceptionCaught", "SwallowedException") e: Throwable,
|
||||||
) {
|
) {
|
||||||
|
|||||||
@@ -0,0 +1,66 @@
|
|||||||
|
package com.fabledsword.minstrel.metadata
|
||||||
|
|
||||||
|
import com.fabledsword.minstrel.cache.db.dao.CachedAlbumDao
|
||||||
|
import com.fabledsword.minstrel.cache.db.dao.CachedArtistDao
|
||||||
|
import com.fabledsword.minstrel.cache.db.dao.CachedTrackDao
|
||||||
|
import com.fabledsword.minstrel.library.data.LibraryRepository
|
||||||
|
import io.mockk.coEvery
|
||||||
|
import io.mockk.coVerify
|
||||||
|
import io.mockk.mockk
|
||||||
|
import kotlinx.coroutines.CompletableDeferred
|
||||||
|
import kotlinx.coroutines.flow.first
|
||||||
|
import kotlinx.coroutines.flow.flowOf
|
||||||
|
import kotlinx.coroutines.launch
|
||||||
|
import kotlinx.coroutines.test.UnconfinedTestDispatcher
|
||||||
|
import kotlinx.coroutines.test.runTest
|
||||||
|
import org.junit.jupiter.api.Test
|
||||||
|
import java.util.concurrent.atomic.AtomicInteger
|
||||||
|
import kotlin.test.assertTrue
|
||||||
|
|
||||||
|
class MetadataProviderTest {
|
||||||
|
private val albumDao = mockk<CachedAlbumDao>(relaxed = true)
|
||||||
|
private val artistDao = mockk<CachedArtistDao>(relaxed = true)
|
||||||
|
private val trackDao = mockk<CachedTrackDao>(relaxed = true)
|
||||||
|
private val library = mockk<LibraryRepository>(relaxed = true)
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `same missing id fetched once (dedup)`() = runTest {
|
||||||
|
val scope = this
|
||||||
|
coEvery { albumDao.observeById("a1") } returns flowOf(null)
|
||||||
|
val provider = MetadataProvider(albumDao, artistDao, trackDao, library, scope)
|
||||||
|
|
||||||
|
provider.observeAlbum("a1").first()
|
||||||
|
provider.observeAlbum("a1").first()
|
||||||
|
testScheduler.advanceUntilIdle()
|
||||||
|
|
||||||
|
coVerify(exactly = 1) { library.refreshAlbumDetail("a1") }
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun `at most four fetches run concurrently`() = runTest {
|
||||||
|
val scope = this
|
||||||
|
val live = AtomicInteger(0)
|
||||||
|
val peak = AtomicInteger(0)
|
||||||
|
val gate = CompletableDeferred<Unit>()
|
||||||
|
coEvery { library.refreshAlbumDetail(any()) } coAnswers {
|
||||||
|
val now = live.incrementAndGet()
|
||||||
|
peak.updateAndGet { p -> maxOf(p, now) }
|
||||||
|
gate.await()
|
||||||
|
live.decrementAndGet()
|
||||||
|
mockk(relaxed = true)
|
||||||
|
}
|
||||||
|
(1..10).forEach { i -> coEvery { albumDao.observeById("a$i") } returns flowOf(null) }
|
||||||
|
val provider = MetadataProvider(albumDao, artistDao, trackDao, library, scope)
|
||||||
|
|
||||||
|
(1..10).forEach { i ->
|
||||||
|
launch(UnconfinedTestDispatcher(testScheduler)) {
|
||||||
|
provider.observeAlbum("a$i").first()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
testScheduler.advanceUntilIdle()
|
||||||
|
|
||||||
|
assertTrue(peak.get() <= 4, "peak concurrency was ${peak.get()}, expected <= 4")
|
||||||
|
gate.complete(Unit)
|
||||||
|
testScheduler.advanceUntilIdle()
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user