mirror of
https://github.com/austinried/subtracks.git
synced 2026-11-11 07:29:14 +01:00
sync: batch the fallback getAlbum song fetches
The `search3`-less fallback emitted one batch per album, and `SyncService.writeSongs` opens a transaction per batch, so syncing a large library that way ran one transaction per album — thousands of fsyncs, and an invalidation round for every list observing `songs` with each. Accumulate what each `getAlbum` returns and flush at `PAGE_SIZE`, so the fallback writes the same ~500-row batches every other fetch already does. Albums are fetched concurrently and complete out of order, so the accumulator sits behind a mutex and the send happens outside it. Batches are in-flight memory, so the flush is per rows reached rather than per album page. Measured per `Song` (strings, boxed longs and genre list included): ~0.7 KiB, so a 500-row batch is ~334 KiB plus ~51 KiB of the `names`/`genres` structures `writeSongs` derives, where a whole album page (~5,500 rows) would be ~3.8 MiB. `albumSongs()` is also a bare `channelFlow`, whose channel holds 64 elements whatever their size, so 500-row batches would have made its worst case ~21 MiB; it now carries the same `.buffer(1)` that `entityBatches()` already uses on itself. The rest of the in-flight rows scale with `syncConcurrency`, so that leaves the default's batch in hand and a few MiB at the highest setting. `SubsonicSourceTest.albumSongsAreEmittedInBoundedBatches` asserts the emitted sizes, [500, 100] for 600 songs spread over 200 three-song albums, which pins the carried remainder as well as the flush: verified it fails on the pre-change code (200 batches of 3) and against a mutant that drops the remainder ([500, 99]). `docs/architecture.md`'s page cap was 20x out of date, so it is corrected alongside. `ktlintCheck`, `testDebugUnitTest` and `lintDebug` pass.
This commit is contained in:
@@ -25,7 +25,9 @@ import kotlinx.coroutines.flow.flow
|
||||
import kotlinx.coroutines.flow.flowOn
|
||||
import kotlinx.coroutines.flow.map
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.sync.Mutex
|
||||
import kotlinx.coroutines.sync.Semaphore
|
||||
import kotlinx.coroutines.sync.withLock
|
||||
import kotlinx.coroutines.sync.withPermit
|
||||
import kotlinx.coroutines.withContext
|
||||
import org.xml.sax.Attributes
|
||||
@@ -224,6 +226,23 @@ class SubsonicSource(
|
||||
private fun albumSongs(): Flow<List<Song>> =
|
||||
channelFlow {
|
||||
val permits = Semaphore(maxConcurrentFetches)
|
||||
val lock = Mutex()
|
||||
val pending = ArrayList<Song>(PAGE_SIZE)
|
||||
|
||||
suspend fun accept(songs: List<Song>) {
|
||||
lock
|
||||
.withLock {
|
||||
pending.addAll(songs)
|
||||
if (pending.size < PAGE_SIZE) {
|
||||
null
|
||||
} else {
|
||||
val batch = ArrayList(pending.subList(0, PAGE_SIZE))
|
||||
pending.subList(0, PAGE_SIZE).clear()
|
||||
batch
|
||||
}
|
||||
}?.let { send(it) }
|
||||
}
|
||||
|
||||
var offset = 0
|
||||
var pages = 0
|
||||
while (true) {
|
||||
@@ -251,7 +270,7 @@ class SubsonicSource(
|
||||
onChild = { song, name, attrs -> if (name == "genres") song.addGenre(attrs) },
|
||||
).collect { batch ->
|
||||
val accepted = batch.map { it.toSong() }.filter { it.id.isNotEmpty() }
|
||||
if (accepted.isNotEmpty()) send(accepted)
|
||||
accept(accepted)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -265,7 +284,14 @@ class SubsonicSource(
|
||||
if (pages > maxPages) throw pageCapExceeded()
|
||||
offset += PAGE_SIZE
|
||||
}
|
||||
}
|
||||
val tail =
|
||||
lock.withLock {
|
||||
val held = pending.toList()
|
||||
pending.clear()
|
||||
held
|
||||
}
|
||||
if (tail.isNotEmpty()) send(tail)
|
||||
}.buffer(1)
|
||||
|
||||
private fun supportsEmptyQuerySearch(): Boolean {
|
||||
emptyQuerySearchSupported?.let { return it }
|
||||
|
||||
Reference in New Issue
Block a user