Skip to content

[MS-949] Introduce Concurrency for Face and Fingerprint Matching #1169

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
meladRaouf merged 1 commit into from
May 28, 2025
Merged

[MS-949] Introduce Concurrency for Face and Fingerprint Matching #1169

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
113 changes: 64 additions & 49 deletions feature/matcher/src/main/java/com/simprints/matcher/usecases/FaceMatcherUseCase.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,32 +4,38 @@ import com.simprints.core.DispatcherBG
import com.simprints.face.infra.basebiosdk.matching.FaceIdentity
import com.simprints.face.infra.basebiosdk.matching.FaceMatcher
import com.simprints.face.infra.basebiosdk.matching.FaceSample
import com.simprints.face.infra.biosdkresolver.FaceBioSDK
import com.simprints.face.infra.biosdkresolver.ResolveFaceBioSdkUseCase
import com.simprints.infra.config.store.models.Project
import com.simprints.infra.enrolment.records.repository.EnrolmentRecordRepository
import com.simprints.infra.enrolment.records.repository.domain.models.BiometricDataSource
import com.simprints.infra.enrolment.records.repository.domain.models.SubjectQuery
import com.simprints.infra.logging.LoggingConstants
import com.simprints.infra.logging.Simber
import com.simprints.matcher.FaceMatchResult
import com.simprints.matcher.MatchParams
import com.simprints.matcher.usecases.MatcherUseCase.MatcherState
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.launch
import javax.inject.Inject
import kotlin.math.min
import com.simprints.infra.enrolment.records.repository.domain.models.FaceIdentity as DomainFaceIdentity

internal class FaceMatcherUseCase @Inject constructor(
private val enrolmentRecordRepository: EnrolmentRecordRepository,
private val resolveFaceBioSdk: ResolveFaceBioSdkUseCase,
private val createRanges: CreateRangesUseCase,
@DispatcherBG private val dispatcher: CoroutineDispatcher,
@DispatcherBG private val dispatcherBG: CoroutineDispatcher,
) : MatcherUseCase {
override val crashReportTag = LoggingConstants.CrashReportTag.FACE_MATCHING

// When using local DB loadedCandidates = expectedCandidates
// However, when using CommCare as data source, loadedCandidates < expectedCandidates
// as it's count function does not take into account filtering criteria
// This var is not thread safe
var loadedCandidates = 0
Copy link

Copilot AI May 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

loadedCandidates is a mutable class-level variable and not thread-safe. Move this counter inside invoke to avoid data races when matching runs in parallel.

Copilot uses AI. Check for mistakes.

override suspend operator fun invoke(
matchParams: MatchParams,
project: Project,
Expand All @@ -53,60 +59,57 @@ internal class FaceMatcherUseCase @Inject constructor(
send(MatcherState.Success(emptyList(), 0, bioSdk.matcherName))
return@channelFlow
}

loadedCandidates = 0
Simber.i("Matching candidates", tag = crashReportTag)
send(MatcherState.LoadingStarted(expectedCandidates))
// When using local DB loadedCandidates = expectedCandidates
// However, when using CommCare as data source, loadedCandidates < expectedCandidates
// as it's count function does not take into account filtering criteria
var loadedCandidates = 0
val resultItems = coroutineScope {
val ranges = createRanges(expectedCandidates)
// if number of ranges less than the number of cores then use the number of ranges
val numConsumers = min(Runtime.getRuntime().availableProcessors(), ranges.size)

createRanges(expectedCandidates)
.map { range ->
async(dispatcher) {
val batchCandidates = getCandidates(
queryWithSupportedFormat,
range,
project = project,
dataSource = matchParams.biometricDataSource,
) {
// When a candidate is loaded
loadedCandidates++
trySend(MatcherState.CandidateLoaded)
}
bioSdk.createMatcher(samples).use { match(it, batchCandidates) }
}
}.awaitAll()
.reduce { acc, subSet -> acc.addAll(subSet) }
.toList()
}
val resultSet = MatchResultSet<FaceMatchResult.Item>()
val candidatesChannel = enrolmentRecordRepository
.loadFaceIdentities(
query = queryWithSupportedFormat,
ranges = ranges,
dataSource = matchParams.biometricDataSource,
project = project,
scope = this,
onCandidateLoaded = {
loadedCandidates++
this.trySend(MatcherState.CandidateLoaded)
},
)

Simber.i("Matched $loadedCandidates candidates", tag = crashReportTag)
// Start Consumers in BG thread
val consumerJobs = List(numConsumers) {
launch(dispatcherBG) {
consumeAndMatch(candidatesChannel, samples, resultSet, bioSdk)
}
}
// Wait for all to complete
consumerJobs.forEach { it.join() }
send(MatcherState.Success(resultSet.toList(), loadedCandidates, bioSdk.matcherName))
}

send(MatcherState.Success(resultItems, loadedCandidates, bioSdk.matcherName))
suspend fun consumeAndMatch(
candidatesChannel: ReceiveChannel<List<DomainFaceIdentity>>,
samples: List<FaceSample>,
resultSet: MatchResultSet<FaceMatchResult.Item>,
bioSdk: FaceBioSDK,
) {
for (batch in candidatesChannel) {
val results = bioSdk.createMatcher(samples).use { matcher ->
match(matcher, batch.mapToFaceIdentities())
}
resultSet.addAll(results)
}
}

private fun mapSamples(probes: List<MatchParams.FaceSample>) = probes.map { FaceSample(it.faceId, it.template) }

private suspend fun getCandidates(
query: SubjectQuery,
range: IntRange,
dataSource: BiometricDataSource = BiometricDataSource.Simprints,
project: Project,
onCandidateLoaded: () -> Unit,
) = enrolmentRecordRepository
.loadFaceIdentities(query, range, dataSource, project, onCandidateLoaded)
.map {
FaceIdentity(
it.subjectId,
it.faces.map { face -> FaceSample(face.id, face.template) },
)
}

private suspend fun match(
matcher: FaceMatcher,
batchCandidates: List<FaceIdentity>
batchCandidates: List<FaceIdentity>,
) = batchCandidates.fold(MatchResultSet<FaceMatchResult.Item>()) { acc, candidate ->
acc.add(
FaceMatchResult.Item(
Expand All @@ -115,4 +118,16 @@ internal class FaceMatcherUseCase @Inject constructor(
),
)
}

private fun List<DomainFaceIdentity>.mapToFaceIdentities(): List<FaceIdentity> = map {
FaceIdentity(
it.subjectId,
it.faces.map {
FaceSample(
it.referenceId,
it.template,
)
},
)
}
}
114 changes: 67 additions & 47 deletions feature/matcher/src/main/java/com/simprints/matcher/usecases/FingerprintMatcherUseCase.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,29 +13,35 @@ import com.simprints.infra.config.store.models.FingerprintConfiguration.FingerCo
import com.simprints.infra.config.store.models.Project
import com.simprints.infra.config.sync.ConfigManager
import com.simprints.infra.enrolment.records.repository.EnrolmentRecordRepository
import com.simprints.infra.enrolment.records.repository.domain.models.BiometricDataSource
import com.simprints.infra.enrolment.records.repository.domain.models.SubjectQuery
import com.simprints.infra.logging.LoggingConstants
import com.simprints.infra.logging.Simber
import com.simprints.matcher.FingerprintMatchResult
import com.simprints.matcher.MatchParams
import com.simprints.matcher.usecases.MatcherUseCase.MatcherState
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.channelFlow
import kotlinx.coroutines.launch
import javax.inject.Inject
import kotlin.math.min
import com.simprints.infra.enrolment.records.repository.domain.models.FingerprintIdentity as DomainFingerprintIdentity

internal class FingerprintMatcherUseCase @Inject constructor(
private val enrolmentRecordRepository: EnrolmentRecordRepository,
private val resolveBioSdkWrapper: ResolveBioSdkWrapperUseCase,
private val configManager: ConfigManager,
private val createRanges: CreateRangesUseCase,
@DispatcherBG private val dispatcher: CoroutineDispatcher,
@DispatcherBG private val dispatcherBG: CoroutineDispatcher,
) : MatcherUseCase {
override val crashReportTag = LoggingConstants.CrashReportTag.FINGER_MATCHING

// When using local DB loadedCandidates = expectedCandidates
// However, when using CommCare as data source, loadedCandidates < expectedCandidates
// as it's count function does not take into account filtering criteria
// This var is not thread safe
var loadedCandidates = 0

Comment on lines +42 to +44
Copy link

Copilot AI May 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

loadedCandidates is a mutable class-level variable and not thread-safe if invoke is called concurrently. Consider making it a local variable inside the invoke method.

Suggested change
// This var is not thread safe
var loadedCandidates = 0

Copilot uses AI. Check for mistakes.
override suspend operator fun invoke(
matchParams: MatchParams,
project: Project,
Expand All @@ -61,55 +67,56 @@ internal class FingerprintMatcherUseCase @Inject constructor(

Simber.i("Matching candidates", tag = crashReportTag)
send(MatcherState.LoadingStarted(expectedCandidates))
// When using local DB loadedCandidates = expectedCandidates
// However, when using CommCare as data source, loadedCandidates < expectedCandidates
// as it's count function does not take into account filtering criteria
var loadedCandidates = 0
val resultItems = createRanges(expectedCandidates)
.map { range ->
async(dispatcher) {
val batchCandidates = getCandidates(queryWithSupportedFormat, range, matchParams.biometricDataSource, project) {
// When a candidate is loaded
loadedCandidates++
trySend(MatcherState.CandidateLoaded)
}
match(samples, batchCandidates, matchParams.flowType, bioSdkWrapper, bioSdk = matchParams.fingerprintSDK)
.fold(MatchResultSet<FingerprintMatchResult.Item>()) { acc, item ->
acc.add(FingerprintMatchResult.Item(item.id, item.score))
}
}
}.awaitAll()
.reduce { acc, subSet -> acc.addAll(subSet) }
.toList()
loadedCandidates = 0
val ranges = createRanges(expectedCandidates)
// if number of ranges less than the number of cores then use the number of ranges
val numConsumers = min(Runtime.getRuntime().availableProcessors(), ranges.size)
val channel = enrolmentRecordRepository.loadFingerprintIdentities(
query = queryWithSupportedFormat,
ranges = ranges,
dataSource = matchParams.biometricDataSource,
scope = this,
project = project,
) {
loadedCandidates++
trySend(MatcherState.CandidateLoaded)
}

val resultSet = MatchResultSet<FingerprintMatchResult.Item>()

// Start Consumers in BG thread
val consumerJobs = List(numConsumers) {
launch(dispatcherBG) {
consumeAndMatch(channel, samples, resultSet, bioSdkWrapper, matchParams)
}
}
// Wait for all to complete
consumerJobs.forEach { it.join() }

Simber.i("Matched $loadedCandidates candidates", tag = crashReportTag)
send(MatcherState.Success(resultItems, loadedCandidates, bioSdkWrapper.matcherName))
send(MatcherState.Success(resultSet.toList(), loadedCandidates, bioSdkWrapper.matcherName))
}

private suspend fun consumeAndMatch(
channel: ReceiveChannel<List<DomainFingerprintIdentity>>,
samples: List<Fingerprint>,
resultSet: MatchResultSet<FingerprintMatchResult.Item>,
bioSdkWrapper: BioSdkWrapper,
matchParams: MatchParams,
) {
for (batch in channel) {
val matchResults =
match(samples, batch.mapToFingerprintIdentity(), matchParams.flowType, bioSdkWrapper, bioSdk = matchParams.fingerprintSDK!!)
.fold(MatchResultSet<FingerprintMatchResult.Item>()) { acc, item ->
acc.add(FingerprintMatchResult.Item(item.id, item.score))
}
resultSet.addAll(matchResults)
}
Comment thread
BurningAXE marked this conversation as resolved.
}

private fun mapSamples(probes: List<MatchParams.FingerprintSample>) = probes
.map { Fingerprint(it.fingerId.toMatcherDomain(), it.template, it.format) }

private suspend fun getCandidates(
query: SubjectQuery,
range: IntRange,
dataSource: BiometricDataSource = BiometricDataSource.Simprints,
project: Project,
onCandidateLoaded: () -> Unit,
) = enrolmentRecordRepository
.loadFingerprintIdentities(query, range, dataSource, project, onCandidateLoaded)
.map {
FingerprintIdentity(
it.subjectId,
it.fingerprints.map { finger ->
Fingerprint(
finger.fingerIdentifier.toMatcherDomain(),
finger.template,
finger.format,
)
},
)
}

private suspend fun match(
probes: List<Fingerprint>,
candidates: List<FingerprintIdentity>,
Expand Down Expand Up @@ -144,4 +151,17 @@ internal class FingerprintMatcherUseCase @Inject constructor(
IFingerIdentifier.LEFT_4TH_FINGER -> FingerIdentifier.LEFT_4TH_FINGER
IFingerIdentifier.LEFT_5TH_FINGER -> FingerIdentifier.LEFT_5TH_FINGER
}

private fun List<DomainFingerprintIdentity>.mapToFingerprintIdentity() = map {
FingerprintIdentity(
it.subjectId,
it.fingerprints.map { finger ->
Fingerprint(
finger.fingerIdentifier.toMatcherDomain(),
finger.template,
finger.format,
)
},
)
}
}
6 changes: 3 additions & 3 deletions feature/matcher/src/test/java/com/simprints/matcher/usecases/FaceMatcherUseCaseTest.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -123,13 +123,13 @@ internal class FaceMatcherUseCaseTest {
)
coEvery { enrolmentRecordRepository.count(any(), any()) } returns 1
coEvery { createRangesUseCase(any()) } returns listOf(0..99)
coEvery { enrolmentRecordRepository.loadFaceIdentities(any(), any(), any(), any(), any()) } coAnswers {
coEvery { enrolmentRecordRepository.loadFaceIdentities(any(), any(), any(), any(), any(), any()) } coAnswers {
// Call the onCandidateLoaded callback (5th parameter)
val onCandidateLoaded = arg<() -> Unit>(4)
val onCandidateLoaded = arg<() -> Unit>(5)
onCandidateLoaded()

// Return the face identities
faceIdentities
createTestChannel(faceIdentities)
}
coEvery { faceMatcher.getHighestComparisonScoreForCandidate(any()) } returns 42f

Expand Down
Loading