From bdf81fe40833c9b4443db745fd4c8c1f84f7194a Mon Sep 17 00:00:00 2001 From: Boof2015 <75185879+Boof2015@users.noreply.github.com> Date: Wed, 12 Aug 2026 13:26:10 -0400 Subject: [PATCH] speedup full scanning --- .../data/LocalScanFastPathTest.kt | 109 ++++++- .../AstraLibraryScannerModule.kt | 303 +++++++++++++----- .../data/AstraLibraryRepository.kt | 143 +++++---- .../data/LocalScanPipeline.kt | 108 +++++++ .../data/LocalScanPipelineTest.kt | 163 ++++++++++ 5 files changed, 668 insertions(+), 158 deletions(-) create mode 100644 modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/LocalScanPipeline.kt create mode 100644 modules/astra-library-scanner/android/src/test/java/expo/modules/astralibraryscanner/data/LocalScanPipelineTest.kt diff --git a/modules/astra-library-scanner/android/src/androidTest/java/expo/modules/astralibraryscanner/data/LocalScanFastPathTest.kt b/modules/astra-library-scanner/android/src/androidTest/java/expo/modules/astralibraryscanner/data/LocalScanFastPathTest.kt index cab6168..f062b81 100644 --- a/modules/astra-library-scanner/android/src/androidTest/java/expo/modules/astralibraryscanner/data/LocalScanFastPathTest.kt +++ b/modules/astra-library-scanner/android/src/androidTest/java/expo/modules/astralibraryscanner/data/LocalScanFastPathTest.kt @@ -4,17 +4,124 @@ import android.content.Context import androidx.test.core.app.ApplicationProvider import androidx.test.ext.junit.runners.AndroidJUnit4 import androidx.test.filters.LargeTest +import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger import kotlinx.coroutines.runBlocking import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse import org.junit.Assert.assertNotEquals import org.junit.Assert.assertNotNull +import org.junit.Assert.assertTrue +import org.junit.Assert.fail import org.junit.Test import org.junit.runner.RunWith @LargeTest @RunWith(AndroidJUnit4::class) class LocalScanFastPathTest { + @Test + fun fullScanFlushesOrderedWindowsAndRetainsCatalogOnCancellationOrFailure() = runBlocking { + val context = ApplicationProvider.getApplicationContext() + val repository = AstraLibraryRepository.get(context) + val suffix = System.nanoTime().toString() + val treeUri = + "content://com.android.externalstorage.documents/tree/primary%3AMusic%2Fpipeline-$suffix" + val parentUri = + "$treeUri/document/primary%3AMusic%2Fpipeline-$suffix" + val files = (0 until 101).map { index -> + file( + uri = "$parentUri%2Ftrack-${index.toString().padStart(3, '0')}.flac", + parentUri = parentUri, + size = 1_024L + index, + mtime = 100L + index, + ) + } + val folder = repository.registerFolder(treeUri, "Pipeline scan test") + val folderId = (folder.getValue("id") as Number).toLong() + + try { + val firstProgress = mutableListOf() + val first = repository.scanLocalFolder( + folderId = folderId, + full = true, + discover = { files }, + extract = { discovered -> metadata(discovered.name) }, + onProgress = { phase, processed, _, _ -> + if (phase == "extracting") firstProgress += processed + }, + ) + assertEquals(101, first.added) + assertEquals(0, first.updated) + assertEquals(0, first.removed) + assertEquals(101, first.total) + assertEquals(listOf(96, 101), firstProgress) + assertTrue(firstProgress.zipWithNext().all { (before, after) -> before < after }) + val firstTitles = files.map { discovered -> + repository.getTrack(discovered.uri)?.get("title") + } + assertEquals(files.map(LocalAudioFile::name), firstTitles) + + val secondProgress = mutableListOf() + val second = repository.scanLocalFolder( + folderId = folderId, + full = true, + discover = { files }, + extract = { discovered -> metadata(discovered.name) }, + onProgress = { phase, processed, _, _ -> + if (phase == "extracting") secondProgress += processed + }, + ) + assertNotEquals(first.revision, second.revision) + assertEquals(0, second.added) + assertEquals(101, second.updated) + assertEquals(0, second.removed) + assertEquals(listOf(96, 101), secondProgress) + assertEquals( + firstTitles, + files.map { discovered -> repository.getTrack(discovered.uri)?.get("title") }, + ) + + val cancelFlag = AtomicBoolean(false) + val extractionCount = AtomicInteger() + val cancelled = repository.scanLocalFolder( + folderId = folderId, + full = true, + discover = { files }, + extract = { discovered -> + if (extractionCount.incrementAndGet() == 3) cancelFlag.set(true) + metadata("Cancelled ${discovered.name}") + }, + onProgress = { _, _, _, _ -> }, + isCancelled = cancelFlag::get, + ) + assertTrue(cancelled.cancelled) + assertEquals(second.revision, cancelled.revision) + assertEquals(second.revision, repository.status().catalogRevision) + assertEquals(firstTitles[0], repository.getTrack(files[0].uri)?.get("title")) + + try { + repository.scanLocalFolder( + folderId = folderId, + full = true, + discover = { files }, + extract = { discovered -> + if (discovered == files[3]) error("parser failure") + metadata("Failed ${discovered.name}") + }, + onProgress = { _, _, _, _ -> }, + ) + fail("Expected parser failure") + } catch (error: IllegalStateException) { + assertEquals("parser failure", error.message) + } + assertEquals(second.revision, repository.status().catalogRevision) + assertEquals(firstTitles[0], repository.getTrack(files[0].uri)?.get("title")) + assertFalse(repository.getTrack(files.last().uri).isNullOrEmpty()) + } finally { + repository.removeFolder(folderId) + } + } + @Test fun unchangedIncrementalScanKeepsRevisionWhileChangesAndFullScansPublish() = runBlocking { val context = ApplicationProvider.getApplicationContext() @@ -125,7 +232,7 @@ class LocalScanFastPathTest { mtime: Long, ): LocalAudioFile = LocalAudioFile( uri = uri, - name = "song.flac", + name = uri.substringAfterLast("%2F").substringAfterLast('/'), size = size, lastModified = mtime, mimeType = "audio/flac", diff --git a/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/AstraLibraryScannerModule.kt b/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/AstraLibraryScannerModule.kt index 4513674..4bf784d 100644 --- a/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/AstraLibraryScannerModule.kt +++ b/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/AstraLibraryScannerModule.kt @@ -2,6 +2,7 @@ package expo.modules.astralibraryscanner import android.content.Context import android.content.Intent +import android.content.pm.ApplicationInfo import android.graphics.Bitmap import android.graphics.BitmapFactory import android.media.AudioFormat @@ -15,7 +16,9 @@ import android.net.Uri import android.os.Build import android.os.Handler import android.os.HandlerThread +import android.os.SystemClock import android.provider.DocumentsContract +import android.util.Log import com.google.android.exoplayer2.MediaItem import com.google.android.exoplayer2.MetadataRetriever import com.google.android.exoplayer2.metadata.id3.BinaryFrame @@ -59,6 +62,86 @@ import java.security.MessageDigest import java.util.concurrent.ConcurrentHashMap import kotlin.math.roundToInt +private const val LIBRARY_SCAN_LOG_TAG = "AstraLibraryScan" + +private data class MetadataStageTiming( + val androidMetadataNanos: Long, + val multiArtistNanos: Long, + val technicalFormatNanos: Long, + val artworkNanos: Long, + val totalNanos: Long, +) + +private data class MetadataTimingSnapshot( + val count: Int, + val androidMetadataNanos: Long, + val multiArtistNanos: Long, + val technicalFormatNanos: Long, + val artworkNanos: Long, + val totalNanos: Long, + val maximumTotalNanos: Long, +) + +private class MetadataTimingAccumulator { + private var count = 0 + private var androidMetadataNanos = 0L + private var multiArtistNanos = 0L + private var technicalFormatNanos = 0L + private var artworkNanos = 0L + private var totalNanos = 0L + private var maximumTotalNanos = 0L + + @Synchronized + fun record(timing: MetadataStageTiming) { + count += 1 + androidMetadataNanos += timing.androidMetadataNanos + multiArtistNanos += timing.multiArtistNanos + technicalFormatNanos += timing.technicalFormatNanos + artworkNanos += timing.artworkNanos + totalNanos += timing.totalNanos + maximumTotalNanos = maxOf(maximumTotalNanos, timing.totalNanos) + } + + @Synchronized + private fun snapshot(): MetadataTimingSnapshot = MetadataTimingSnapshot( + count = count, + androidMetadataNanos = androidMetadataNanos, + multiArtistNanos = multiArtistNanos, + technicalFormatNanos = technicalFormatNanos, + artworkNanos = artworkNanos, + totalNanos = totalNanos, + maximumTotalNanos = maximumTotalNanos, + ) + + fun logIfDebuggable(context: Context, folderId: Long) { + val debuggable = + context.applicationInfo.flags and ApplicationInfo.FLAG_DEBUGGABLE != 0 + if (!debuggable) return + val timing = snapshot() + Log.d( + LIBRARY_SCAN_LOG_TAG, + "folderId=$folderId metadataFiles=${timing.count}" + + " androidMetadataTotalMs=${nanosToMs(timing.androidMetadataNanos)}" + + " androidMetadataAvgMs=${averageMs(timing.androidMetadataNanos, timing.count)}" + + " multiArtistTotalMs=${nanosToMs(timing.multiArtistNanos)}" + + " multiArtistAvgMs=${averageMs(timing.multiArtistNanos, timing.count)}" + + " technicalFormatTotalMs=${nanosToMs(timing.technicalFormatNanos)}" + + " technicalFormatAvgMs=${averageMs(timing.technicalFormatNanos, timing.count)}" + + " artworkTotalMs=${nanosToMs(timing.artworkNanos)}" + + " artworkAvgMs=${averageMs(timing.artworkNanos, timing.count)}" + + " trackExtractionTotalMs=${nanosToMs(timing.totalNanos)}" + + " trackExtractionAvgMs=${averageMs(timing.totalNanos, timing.count)}" + + " trackExtractionMaxMs=${nanosToMs(timing.maximumTotalNanos)}", + ) + } + + private fun averageMs(totalNanos: Long, count: Int): Double = + if (count == 0) 0.0 else nanosToMs(totalNanos / count) + + private fun nanosToMs(nanos: Long): Double = + (nanos / 10_000.0).roundToInt() / 100.0 +} + class FileRequest : Record { @Field val uri: String = "" @Field val coverUri: String? = null @@ -143,11 +226,13 @@ class AstraLibraryScannerModule : Module() { mode: String, extensions: List, -> + val context = requireContext().applicationContext val cancelFlag = AtomicBoolean(false) + val metadataTimings = MetadataTimingAccumulator() activeScans.add(cancelFlag) try { withContext(Dispatchers.IO) { - val repository = AstraLibraryRepository.get(requireContext()) + val repository = AstraLibraryRepository.get(context) repository.withUserRecovery { scanLocalFolder( folderId = folderId.toLong(), full = mode == "full", @@ -172,7 +257,7 @@ class AstraLibraryScannerModule : Module() { } }, extract = { file -> - extractOne(file.uri, file.coverUri).toLocalAudioMetadata() + extractOne(file.uri, file.coverUri, metadataTimings::record).toLocalAudioMetadata() }, onProgress = { phase, processed, total, folderName -> sendEvent( @@ -189,6 +274,7 @@ class AstraLibraryScannerModule : Module() { ).toMap() } } } finally { + runCatching { metadataTimings.logIfDebuggable(context, folderId.toLong()) } activeScans.remove(cancelFlag) } } @@ -625,104 +711,145 @@ class AstraLibraryScannerModule : Module() { return extractOne(request.uri, request.coverUri) } - private fun extractOne(uriString: String, coverUri: String?): Map { + private fun extractOne( + uriString: String, + coverUri: String?, + timingRecorder: ((MetadataStageTiming) -> Unit)? = null, + ): Map { + val totalStartedNanos = SystemClock.elapsedRealtimeNanos() + var androidMetadataNanos = 0L + var multiArtistNanos = 0L + var technicalFormatNanos = 0L + var artworkNanos = 0L val context = requireContext() val uri = Uri.parse(uriString) val result = mutableMapOf("uri" to uriString, "ok" to true) - var embeddedPicture: ByteArray? = null - val retriever = MediaMetadataRetriever() try { - retriever.setDataSource(context, uri) - - result["title"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_TITLE) - result["artist"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ARTIST) - result["album"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUM) - result["albumArtist"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUMARTIST) - result["genre"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_GENRE) - result["mimeType"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_MIMETYPE) - result["durationMs"] = - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DURATION)?.toLongOrNull() - result["bitrate"] = - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_BITRATE)?.toIntOrNull() - result["trackNumber"] = parseTagNumber( - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_CD_TRACK_NUMBER) - ) - result["discNumber"] = parseTagNumber( - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DISC_NUMBER) - ) - result["year"] = parseYear( - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_YEAR), - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DATE) - ) - if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) { - result["sampleRate"] = - retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_SAMPLERATE)?.toIntOrNull() - } - embeddedPicture = retriever.embeddedPicture - } catch (t: Throwable) { - return mapOf( - "uri" to uriString, - "ok" to false, - "error" to (t.message ?: t.javaClass.simpleName) - ) - } finally { + var embeddedPicture: ByteArray? = null + var androidMetadataError: Throwable? = null + val androidMetadataStartedNanos = SystemClock.elapsedRealtimeNanos() + val retriever = MediaMetadataRetriever() try { - retriever.release() - } catch (_: Throwable) {} - } + retriever.setDataSource(context, uri) - val credits = ArtistCreditMetadataReader.read(context, uri, metadataTimeoutMs) - val artistNames = credits.artists.takeIf { it.size > 1 }.orEmpty() - val albumArtistNames = credits.albumArtists.takeIf { it.size > 1 }.orEmpty() - result["artistNames"] = artistNames - result["albumArtistNames"] = albumArtistNames - if (artistNames.isNotEmpty()) { - result["artist"] = formatArtistNames(artistNames) - } else if (result["artist"] == null && credits.artists.size == 1) { - result["artist"] = credits.artists[0] - } - if (albumArtistNames.isNotEmpty()) { - result["albumArtist"] = formatArtistNames(albumArtistNames) - } else if (result["albumArtist"] == null && credits.albumArtists.size == 1) { - result["albumArtist"] = credits.albumArtists[0] - } - - // Header-level facts MMR can't provide (channels, bit depth) or only on - // API 31+ (sample rate). Failure here is non-fatal — keep the tag data. - val extractor = MediaExtractor() - try { - extractor.setDataSource(context, uri, null) - for (i in 0 until extractor.trackCount) { - val format = extractor.getTrackFormat(i) - val trackMime = format.getString(MediaFormat.KEY_MIME) ?: continue - if (!trackMime.startsWith("audio/")) continue - - result["codecMime"] = trackMime - if (format.containsKey(MediaFormat.KEY_CHANNEL_COUNT)) { - result["channels"] = format.getInteger(MediaFormat.KEY_CHANNEL_COUNT) + result["title"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_TITLE) + result["artist"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ARTIST) + result["album"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUM) + result["albumArtist"] = + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUMARTIST) + result["genre"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_GENRE) + result["mimeType"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_MIMETYPE) + result["durationMs"] = + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DURATION)?.toLongOrNull() + result["bitrate"] = + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_BITRATE)?.toIntOrNull() + result["trackNumber"] = parseTagNumber( + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_CD_TRACK_NUMBER) + ) + result["discNumber"] = parseTagNumber( + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DISC_NUMBER) + ) + result["year"] = parseYear( + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_YEAR), + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DATE) + ) + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) { + result["sampleRate"] = + retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_SAMPLERATE)?.toIntOrNull() } - if (result["sampleRate"] == null && format.containsKey(MediaFormat.KEY_SAMPLE_RATE)) { - result["sampleRate"] = format.getInteger(MediaFormat.KEY_SAMPLE_RATE) - } - result["bitsPerSample"] = readBitsPerSample(format) - break + embeddedPicture = retriever.embeddedPicture + } catch (error: Throwable) { + androidMetadataError = error + } finally { + try { + retriever.release() + } catch (_: Throwable) {} + androidMetadataNanos = SystemClock.elapsedRealtimeNanos() - androidMetadataStartedNanos } - } catch (_: Throwable) { - // Container not supported by MediaExtractor — tag data already collected. - } finally { + androidMetadataError?.let { error -> + return mapOf( + "uri" to uriString, + "ok" to false, + "error" to (error.message ?: error.javaClass.simpleName), + ) + } + + val multiArtistStartedNanos = SystemClock.elapsedRealtimeNanos() + val credits = try { + ArtistCreditMetadataReader.read(context, uri, metadataTimeoutMs) + } finally { + multiArtistNanos = SystemClock.elapsedRealtimeNanos() - multiArtistStartedNanos + } + val artistNames = credits.artists.takeIf { it.size > 1 }.orEmpty() + val albumArtistNames = credits.albumArtists.takeIf { it.size > 1 }.orEmpty() + result["artistNames"] = artistNames + result["albumArtistNames"] = albumArtistNames + if (artistNames.isNotEmpty()) { + result["artist"] = formatArtistNames(artistNames) + } else if (result["artist"] == null && credits.artists.size == 1) { + result["artist"] = credits.artists[0] + } + if (albumArtistNames.isNotEmpty()) { + result["albumArtist"] = formatArtistNames(albumArtistNames) + } else if (result["albumArtist"] == null && credits.albumArtists.size == 1) { + result["albumArtist"] = credits.albumArtists[0] + } + + // Header-level facts MMR can't provide (channels, bit depth) or only on + // API 31+ (sample rate). Failure here is non-fatal — keep the tag data. + val technicalFormatStartedNanos = SystemClock.elapsedRealtimeNanos() + val extractor = MediaExtractor() try { - extractor.release() - } catch (_: Throwable) {} - } + extractor.setDataSource(context, uri, null) + for (i in 0 until extractor.trackCount) { + val format = extractor.getTrackFormat(i) + val trackMime = format.getString(MediaFormat.KEY_MIME) ?: continue + if (!trackMime.startsWith("audio/")) continue - try { - result["artworkHash"] = resolveArtwork(embeddedPicture, coverUri) - } catch (_: Throwable) { - // Artwork failure never fails the track. - } + result["codecMime"] = trackMime + if (format.containsKey(MediaFormat.KEY_CHANNEL_COUNT)) { + result["channels"] = format.getInteger(MediaFormat.KEY_CHANNEL_COUNT) + } + if (result["sampleRate"] == null && format.containsKey(MediaFormat.KEY_SAMPLE_RATE)) { + result["sampleRate"] = format.getInteger(MediaFormat.KEY_SAMPLE_RATE) + } + result["bitsPerSample"] = readBitsPerSample(format) + break + } + } catch (_: Throwable) { + // Container not supported by MediaExtractor — tag data already collected. + } finally { + try { + extractor.release() + } catch (_: Throwable) {} + technicalFormatNanos = + SystemClock.elapsedRealtimeNanos() - technicalFormatStartedNanos + } - return result + val artworkStartedNanos = SystemClock.elapsedRealtimeNanos() + try { + result["artworkHash"] = resolveArtwork(embeddedPicture, coverUri) + } catch (_: Throwable) { + // Artwork failure never fails the track. + } finally { + artworkNanos = SystemClock.elapsedRealtimeNanos() - artworkStartedNanos + } + + return result + } finally { + runCatching { + timingRecorder?.invoke( + MetadataStageTiming( + androidMetadataNanos = androidMetadataNanos, + multiArtistNanos = multiArtistNanos, + technicalFormatNanos = technicalFormatNanos, + artworkNanos = artworkNanos, + totalNanos = SystemClock.elapsedRealtimeNanos() - totalStartedNanos, + ), + ) + } + } } private fun Map.toLocalAudioMetadata(): LocalAudioMetadata = diff --git a/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/AstraLibraryRepository.kt b/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/AstraLibraryRepository.kt index ee8b44a..999d51a 100644 --- a/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/AstraLibraryRepository.kt +++ b/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/AstraLibraryRepository.kt @@ -20,9 +20,6 @@ import java.util.concurrent.ConcurrentHashMap import java.util.concurrent.CopyOnWriteArraySet import java.util.concurrent.atomic.AtomicInteger import kotlin.random.Random -import kotlinx.coroutines.async -import kotlinx.coroutines.awaitAll -import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job @@ -62,6 +59,8 @@ private class LocalScanTiming( var readModelMs: Long = 0 var publishMs: Long = 0 var files: Int = 0 + var workers: Int = 0 + var windowSize: Int = 0 var outcome: String = "failed" fun beginLockWait() { @@ -84,6 +83,7 @@ private class LocalScanTiming( Log.d( SCAN_LOG_TAG, "folderId=$folderId outcome=$outcome files=$files" + + " workers=$workers windowSize=$windowSize" + " lockWaitMs=$lockWaitMs discoveryMs=$discoveryMs" + " comparisonMs=$comparisonMs extractionAndStagingMs=$extractionAndStagingMs" + " readModelMs=$readModelMs publishMs=$publishMs" + @@ -552,75 +552,80 @@ class AstraLibraryRepository private constructor( var added = 0 var updated = 0 var errors = 0 - var processed = 0 + val workerCount = localScanWorkerCount( + availableProcessors = Runtime.getRuntime().availableProcessors(), + itemCount = files.size, + ) + timing.workers = workerCount + timing.windowSize = LOCAL_SCAN_WINDOW_SIZE - for (batch in files.chunked(24)) { - throwIfScanCancelled(isCancelled) - val rows = coroutineScope { - batch.map { file -> - async(Dispatchers.IO) { - throwIfScanCancelled(isCancelled) - val old = existingByPath[file.uri] - val unchanged = !effectiveFull && - old != null && - old.mtime == file.lastModified && - old.size == file.size - if (unchanged) { - StagedLocalTrack( - row = old!!.copy( - id = 0, - generationId = generationId, - sourceKey = sourceKey, - titleSortKey = SortKeys.forText(old.title), - artistSortKey = SortKeys.forText(old.artist), - albumSortKey = SortKeys.forText(old.album), - fileNameSortKey = SortKeys.forText(old.fileName), - sectionLabel = SortKeys.sectionLabel(old.title), - ), - failed = false, - metadataChanged = false, - ) - } else { - val metadata = extract(file) - throwIfScanCancelled(isCancelled) - if (!metadata.ok) { - StagedLocalTrack( - row = old?.copy(id = 0, generationId = generationId, sourceKey = sourceKey), - failed = true, - metadataChanged = false, - ) - } else { - StagedLocalTrack( - row = trackFromMetadata( - generationId = generationId, - sourceKey = sourceKey, - folderId = folderId, - file = file, - metadata = metadata, - addedAt = old?.addedAt ?: startedAt, - ), - failed = false, - metadataChanged = true, - ) - } - } + runBoundedLocalScanPipeline( + items = files, + workerCount = workerCount, + process = { file -> + throwIfScanCancelled(isCancelled) + val old = existingByPath[file.uri] + val unchanged = !effectiveFull && + old != null && + old.mtime == file.lastModified && + old.size == file.size + if (unchanged) { + StagedLocalTrack( + row = old!!.copy( + id = 0, + generationId = generationId, + sourceKey = sourceKey, + titleSortKey = SortKeys.forText(old.title), + artistSortKey = SortKeys.forText(old.artist), + albumSortKey = SortKeys.forText(old.album), + fileNameSortKey = SortKeys.forText(old.fileName), + sectionLabel = SortKeys.sectionLabel(old.title), + ), + failed = false, + metadataChanged = false, + ) + } else { + val metadata = extract(file) + throwIfScanCancelled(isCancelled) + if (!metadata.ok) { + StagedLocalTrack( + row = old?.copy(id = 0, generationId = generationId, sourceKey = sourceKey), + failed = true, + metadataChanged = false, + ) + } else { + StagedLocalTrack( + row = trackFromMetadata( + generationId = generationId, + sourceKey = sourceKey, + folderId = folderId, + file = file, + metadata = metadata, + addedAt = old?.addedAt ?: startedAt, + ), + failed = false, + metadataChanged = true, + ) } - }.awaitAll() - } - throwIfScanCancelled(isCancelled) - val insertRows = ArrayList(rows.size) - for (staged in rows) { - if (staged.failed) errors += 1 - val row = staged.row ?: continue - insertRows += row - if (staged.metadataChanged) { - if (existingByPath.containsKey(row.path)) updated += 1 else added += 1 } - } - if (insertRows.isNotEmpty()) dao.putTracks(insertRows) - processed += batch.size - onProgress("extracting", processed, files.size, folder.displayName) - } + }, + writeWindow = { rows -> + throwIfScanCancelled(isCancelled) + val insertRows = ArrayList(rows.size) + for (staged in rows) { + if (staged.failed) errors += 1 + val row = staged.row ?: continue + insertRows += row + if (staged.metadataChanged) { + if (existingByPath.containsKey(row.path)) updated += 1 else added += 1 + } + } + if (insertRows.isNotEmpty()) dao.putTracks(insertRows) + }, + onWindowCommitted = { processed, total -> + onProgress("extracting", processed, total, folder.displayName) + }, + ) timing.extractionAndStagingMs = timing.elapsedMs(extractionAndStagingStarted) throwIfScanCancelled(isCancelled) diff --git a/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/LocalScanPipeline.kt b/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/LocalScanPipeline.kt new file mode 100644 index 0000000..40e53b4 --- /dev/null +++ b/modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/LocalScanPipeline.kt @@ -0,0 +1,108 @@ +package expo.modules.astralibraryscanner.data + +import java.util.concurrent.atomic.AtomicInteger +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.joinAll +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Semaphore + +internal const val LOCAL_SCAN_WINDOW_SIZE = 96 +private const val LOCAL_SCAN_WORKERS_PER_PROCESSOR = 3 +private const val LOCAL_SCAN_MIN_WORKERS = 8 +private const val LOCAL_SCAN_MAX_WORKERS = 24 +private const val LOCAL_SCAN_CHANNEL_CAPACITY = 1 +private const val LOCAL_SCAN_RETAINED_WINDOWS = LOCAL_SCAN_CHANNEL_CAPACITY + 1 + +internal fun localScanWorkerCount( + availableProcessors: Int, + itemCount: Int, +): Int { + if (itemCount <= 0) return 0 + val processors = availableProcessors.coerceAtLeast(1) + val target = (processors * LOCAL_SCAN_WORKERS_PER_PROCESSOR) + .coerceIn(LOCAL_SCAN_MIN_WORKERS, LOCAL_SCAN_MAX_WORKERS) + return minOf(target, itemCount) +} + +/** + * Extracts ordered windows in parallel while a single consumer commits them. + * + * The capacity-one channel plus the two-permit window guard allows one window to be + * written while the next is extracted, without starting a third retained window. + */ +internal suspend fun runBoundedLocalScanPipeline( + items: List, + workerCount: Int, + windowSize: Int = LOCAL_SCAN_WINDOW_SIZE, + process: suspend (Input) -> Output, + writeWindow: suspend (List) -> Unit, + onWindowCommitted: (processed: Int, total: Int) -> Unit = { _, _ -> }, +) { + if (items.isEmpty()) return + require(workerCount > 0) { "workerCount must be positive" } + require(windowSize > 0) { "windowSize must be positive" } + + coroutineScope { + val completedWindows = Channel>(capacity = LOCAL_SCAN_CHANNEL_CAPACITY) + val retainedWindowSlots = Semaphore(LOCAL_SCAN_RETAINED_WINDOWS) + val producer = launch(Dispatchers.IO) { + try { + for (inputWindow in items.chunked(windowSize)) { + currentCoroutineContext().ensureActive() + retainedWindowSlots.acquire() + var handedToWriter = false + try { + val outputWindow = processOrderedWindow(inputWindow, workerCount, process) + completedWindows.send(outputWindow) + handedToWriter = true + } finally { + if (!handedToWriter) retainedWindowSlots.release() + } + } + } finally { + completedWindows.close() + } + } + + var processed = 0 + try { + for (window in completedWindows) { + try { + writeWindow(window) + } finally { + retainedWindowSlots.release() + } + processed += window.size + onWindowCommitted(processed, items.size) + } + producer.join() + } finally { + completedWindows.cancel() + } + } +} + +private suspend fun processOrderedWindow( + items: List, + requestedWorkerCount: Int, + process: suspend (Input) -> Output, +): List = coroutineScope { + val nextIndex = AtomicInteger() + val results = MutableList(items.size) { null } + val workers = List(minOf(requestedWorkerCount, items.size)) { + launch { + while (true) { + currentCoroutineContext().ensureActive() + val index = nextIndex.getAndIncrement() + if (index >= items.size) break + results[index] = process(items[index]) + } + } + } + workers.joinAll() + results.map { checkNotNull(it) } +} diff --git a/modules/astra-library-scanner/android/src/test/java/expo/modules/astralibraryscanner/data/LocalScanPipelineTest.kt b/modules/astra-library-scanner/android/src/test/java/expo/modules/astralibraryscanner/data/LocalScanPipelineTest.kt new file mode 100644 index 0000000..a7dd2a8 --- /dev/null +++ b/modules/astra-library-scanner/android/src/test/java/expo/modules/astralibraryscanner/data/LocalScanPipelineTest.kt @@ -0,0 +1,163 @@ +package expo.modules.astralibraryscanner.data + +import java.util.Collections +import java.util.concurrent.atomic.AtomicInteger +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Assert.fail +import org.junit.Test + +class LocalScanPipelineTest { + @Test + fun workerCountClampsToProcessorsLimitsAndItems() { + assertEquals(0, localScanWorkerCount(16, 0)) + assertEquals(1, localScanWorkerCount(16, 1)) + assertEquals(8, localScanWorkerCount(1, 20)) + assertEquals(12, localScanWorkerCount(4, 20)) + assertEquals(20, localScanWorkerCount(32, 20)) + assertEquals(24, localScanWorkerCount(32, 100)) + assertEquals(3, localScanWorkerCount(8, 3)) + } + + @Test + fun outputStaysOrderedAndWritesFullAndPartialWindows() = runBlocking { + val writes = mutableListOf>() + val progress = mutableListOf() + + runBoundedLocalScanPipeline( + items = (0 until 205).toList(), + workerCount = 8, + process = { value -> + delay(((7 - value % 8) + 1).toLong()) + value + }, + writeWindow = { writes += it }, + onWindowCommitted = { processed, total -> + assertEquals(205, total) + progress += processed + }, + ) + + assertEquals(listOf(96, 96, 13), writes.map(List::size)) + assertEquals((0 until 205).toList(), writes.flatten()) + assertEquals(listOf(96, 192, 205), progress) + } + + @Test + fun activeExtractionNeverExceedsWorkerCount() = runBlocking { + val active = AtomicInteger() + val maximum = AtomicInteger() + + runBoundedLocalScanPipeline( + items = (0 until 24).toList(), + workerCount = 3, + windowSize = 24, + process = { value -> + val current = active.incrementAndGet() + maximum.getAndUpdate { previous -> maxOf(previous, current) } + try { + delay(10) + value + } finally { + active.decrementAndGet() + } + }, + writeWindow = {}, + ) + + assertEquals(3, maximum.get()) + } + + @Test + fun extractionOfNextWindowOverlapsCurrentWrite() = runBlocking { + val secondWindowStarted = CompletableDeferred() + val thirdWindowStarted = CompletableDeferred() + val writes = mutableListOf>() + + withTimeout(2_000) { + runBoundedLocalScanPipeline( + items = (0 until 6).toList(), + workerCount = 2, + windowSize = 2, + process = { value -> + if (value >= 2) secondWindowStarted.complete(Unit) + if (value >= 4) thirdWindowStarted.complete(Unit) + value + }, + writeWindow = { window -> + if (writes.isEmpty()) { + secondWindowStarted.await() + delay(50) + assertFalse(thirdWindowStarted.isCompleted) + } + writes += window + }, + ) + } + + assertEquals(listOf(listOf(0, 1), listOf(2, 3), listOf(4, 5)), writes) + } + + @Test + fun extractionFailureCancelsPipelineAndSkipsUncommittedWindows() = runBlocking { + val processed = Collections.synchronizedList(mutableListOf()) + val writes = mutableListOf>() + + try { + runBoundedLocalScanPipeline( + items = (0 until 12).toList(), + workerCount = 4, + windowSize = 6, + process = { value -> + processed += value + if (value == 3) error("parser failure") + delay(20) + value + }, + writeWindow = { writes += it }, + ) + fail("Expected parser failure") + } catch (error: IllegalStateException) { + assertEquals("parser failure", error.message) + } + + assertTrue(writes.isEmpty()) + assertTrue(processed.size < 12) + } + + @Test + fun writerFailureCancelsFurtherExtraction() = runBlocking { + val active = AtomicInteger() + val completed = AtomicInteger() + + try { + runBoundedLocalScanPipeline( + items = (0 until 30).toList(), + workerCount = 2, + windowSize = 5, + process = { value -> + active.incrementAndGet() + try { + delay(15) + completed.incrementAndGet() + value + } finally { + active.decrementAndGet() + } + }, + writeWindow = { error("database failure") }, + ) + fail("Expected database failure") + } catch (error: IllegalStateException) { + assertEquals("database failure", error.message) + } + + assertEquals(0, active.get()) + assertTrue(completed.get() < 30) + } +}