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

musikr.pipeline: parallelize cache writes

Browse source
This commit is contained in:
Alexander Capehart 2024-12-20 22:21:24 -05:00
parent 2842bd57b1
commit c4a4b69cd1
No known key found for this signature in database
GPG key ID: 37DBE3621FE9AD47
1 changed files with 15 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -73,9 +73,9 @@ private class ExtractStepImpl(
val audioNodes = filterFlow.right val audioNodes = filterFlow.right
val playlistNodes = filterFlow.left.map { ExtractedMusic.Playlist(it) } val playlistNodes = filterFlow.left.map { ExtractedMusic.Playlist(it) }
val distributedAudioNodes = audioNodes.distribute(8) val readDistributedFlow = audioNodes.distribute(8)
val cacheResults = val cacheResults =
distributedAudioNodes.flows readDistributedFlow.flows
.map { flow -> .map { flow ->
flow flow
.map { wrap(it) { file -> cache.read(file, storedCovers) } } .map { wrap(it) { file -> cache.read(file, storedCovers) } }
@ -132,19 +132,26 @@ private class ExtractStepImpl(
.flowOn(Dispatchers.IO) .flowOn(Dispatchers.IO)
.buffer(Channel.UNLIMITED) .buffer(Channel.UNLIMITED)
val writeDistributedFlow = extractedSongs.distribute(8)
val writtenSongs = val writtenSongs =
extractedSongs writeDistributedFlow.flows
.map { .map { flow ->
wrap(it, cache::write) flow
ExtractedMusic.Song(it) .map {
wrap(it, cache::write)
ExtractedMusic.Song(it)
}
.flowOn(Dispatchers.IO)
.buffer(Channel.UNLIMITED)
} }
.flowOn(Dispatchers.IO) .flattenMerge()
return merge( return merge(
filterFlow.manager, filterFlow.manager,
distributedAudioNodes.manager, readDistributedFlow.manager,
cacheFlow.manager, cacheFlow.manager,
cachedSongs, cachedSongs,
writeDistributedFlow.manager,
writtenSongs, writtenSongs,
playlistNodes) playlistNodes)
} }