Fabio/Auxio
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

musikr: parallelize all extraction

Browse source
This commit is contained in:
Alexander Capehart 2025-02-25 16:11:09 -07:00
parent 0387400a4a
commit 584af83a07
No known key found for this signature in database
GPG key ID: 37DBE3621FE9AD47
1 changed files with 61 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
musikr/src/main/java/org/oxycblt/musikr/pipeline/ExtractStep.kt
View file

@ -78,21 +78,43 @@ private class ExtractStepImpl(
val audioNodes = filterFlow.right val audioNodes = filterFlow.right
val playlistNodes = filterFlow.left.map { ExtractedMusic.Valid.Playlist(it) } val playlistNodes = filterFlow.left.map { ExtractedMusic.Valid.Playlist(it) }
// Distribute audio nodes for parallel processing // First distribute audio nodes for parallel cache reading
val processingDistributedFlow = audioNodes.distribute(8) val readDistributedFlow = audioNodes.distribute(8)
val cacheResults =
readDistributedFlow.flows
.map { flow ->
flow
.map { wrap(it) { file -> cache.read(file, storedCovers) } }
.flowOn(Dispatchers.IO)
.buffer(Channel.UNLIMITED)
}
.flattenMerge()
.buffer(Channel.UNLIMITED)
// Process each audio file in parallel flows // Divert cache hits and misses
val cacheFlow =
cacheResults.divert {
when (it) {
is CacheResult.Hit -> Divert.Left(it.song)
is CacheResult.Miss -> Divert.Right(it.file)
}
}
// Cache hits can be directly converted to valid songs
val cachedSongs = cacheFlow.left.map { ExtractedMusic.Valid.Song(it) }
// Process uncached files in parallel
val uncachedFiles = cacheFlow.right
val processingDistributedFlow = uncachedFiles.distribute(8)
// Process each uncached file in parallel flows
val processedSongs = val processedSongs =
processingDistributedFlow.flows processingDistributedFlow.flows
.map { flow -> .map { flow ->
flow flow
.mapNotNull { file -> .mapNotNull { file ->
// First try to read from cache
wrap(file) { f -> wrap(file) { f ->
when (val result = cache.read(f, storedCovers)) { // Open file descriptor
is CacheResult.Hit -> ExtractedMusic.Valid.Song(result.song)
is CacheResult.Miss -> {
// If not in cache, process the file
val fd = withContext(Dispatchers.IO) { val fd = withContext(Dispatchers.IO) {
context.contentResolver.openFileDescriptor(f.uri, "r") context.contentResolver.openFileDescriptor(f.uri, "r")
} ?: return@wrap null } ?: return@wrap null
@ -123,32 +145,33 @@ private class ExtractStepImpl(
} }
} }
} }
}
}
.flowOn(Dispatchers.IO) .flowOn(Dispatchers.IO)
.buffer(Channel.UNLIMITED) .buffer(Channel.UNLIMITED)
} }
.flattenMerge() .flattenMerge()
.buffer(Channel.UNLIMITED) .buffer(Channel.UNLIMITED)
// Separate valid songs from invalid ones // Separate valid processed songs from invalid ones
val processedFlow = processedSongs.divert { val processedFlow = processedSongs.divert {
when (it) { when (it) {
is ExtractedMusic.Valid.Song -> Divert.Left(it) is ExtractedMusic.Valid.Song -> Divert.Left(it)
is ExtractedMusic.Invalid -> Divert.Right(it) is ExtractedMusic.Invalid -> Divert.Right(it)
else -> Divert.Right(ExtractedMusic.Invalid) // Should never happen else -> Divert.Right(ExtractedMusic.Invalid)
} }
} }
val validSongs = processedFlow.left val processedValidSongs = processedFlow.left
val invalidSongs = processedFlow.right val invalidSongs = processedFlow.right
val merged = val merged =
merge( merge(
filterFlow.manager, filterFlow.manager,
readDistributedFlow.manager,
cacheFlow.manager,
processingDistributedFlow.manager, processingDistributedFlow.manager,
processedFlow.manager, processedFlow.manager,
validSongs, cachedSongs,
processedValidSongs,
invalidSongs, invalidSongs,
playlistNodes) playlistNodes)