Skip to content
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

Herald supports reach-only llv2 protocol. #1135

Merged
merged 11 commits into from
Aug 3, 2023
Merged

Herald supports reach-only llv2 protocol. #1135

Show file tree
Hide file tree
Changes from 7 commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
4f1ce7e
Duchy Apis and System Apis support Reach-Only Llv2 protocol.
renjiezh Jul 20, 2023
5ab24ec
Bug fix
renjiezh Jul 20, 2023
2ffd289
Update public api proto.
renjiezh Jul 20, 2023
6ca458d
Merge branch 'main' of github.com:world-federation-of-advertisers/cro…
renjiezh Jul 24, 2023
0726d59
Herald supports reach-only llv2 protocol.
renjiezh Jul 25, 2023
4f3c277
Merge origin main
renjiezh Aug 1, 2023
ab531e4
Update based on PR comments.
renjiezh Aug 1, 2023
9085efc
Update by PR comments.
renjiezh Aug 1, 2023
fde751c
Update by PR comments
renjiezh Aug 2, 2023
b828fcc
Add new line
renjiezh Aug 3, 2023
bfd2358
Merge branch 'main' into renjiez-rollv2-herald
renjiezh Aug 3, 2023
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
4 changes: 4 additions & 0 deletions src/main/k8s/testing/secretfiles/aggregator_protocols_setup_config.textproto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,7 @@ liquid_legions_v2 {
role: AGGREGATOR
external_aggregator_duchy_id: "aggregator"
}
reach_only_liquid_legions_v2 {
role: AGGREGATOR
external_aggregator_duchy_id: "aggregator"
}
4 changes: 4 additions & 0 deletions src/main/k8s/testing/secretfiles/non_aggregator_protocols_setup_config.textproto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,3 +4,7 @@ liquid_legions_v2 {
role: NON_AGGREGATOR
external_aggregator_duchy_id: "aggregator"
}
reach_only_liquid_legions_v2 {
role: NON_AGGREGATOR
external_aggregator_duchy_id: "aggregator"
}
1 change: 1 addition & 0 deletions src/main/kotlin/org/wfanet/measurement/duchy/daemon/herald/BUILD.bazel
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@ kt_jvm_library(
srcs = [
"Herald.kt",
"LiquidLegionsV2Starter.kt",
"ReachOnlyLiquidLegionsV2Starter.kt",
],
runtime_deps = ["@wfa_common_jvm//imports/java/io/grpc/netty"],
deps = [
Expand Down
33 changes: 25 additions & 8 deletions src/main/kotlin/org/wfanet/measurement/duchy/daemon/herald/Herald.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,9 +32,7 @@ import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Semaphore
import org.wfanet.measurement.common.grpc.grpcStatusCode
import org.wfanet.measurement.common.protoTimestamp
import org.wfanet.measurement.duchy.daemon.utils.MeasurementType
import org.wfanet.measurement.duchy.daemon.utils.key
import org.wfanet.measurement.duchy.daemon.utils.toMeasurementType
import org.wfanet.measurement.duchy.service.internal.computations.toGetTokenRequest
import org.wfanet.measurement.internal.duchy.ComputationDetails
import org.wfanet.measurement.internal.duchy.ComputationsGrpcKt.ComputationsCoroutineStub
Expand Down Expand Up @@ -229,16 +227,22 @@ class Herald(
val globalId: String = systemComputation.key.computationId
logger.info("[id=$globalId] Creating Computation...")
try {
when (systemComputation.toMeasurementType()) {
MeasurementType.REACH,
MeasurementType.REACH_AND_FREQUENCY -> {
when (systemComputation.mpcProtocolConfig.protocolCase) {
Computation.MpcProtocolConfig.ProtocolCase.LIQUID_LEGIONS_V2 ->
LiquidLegionsV2Starter.createComputation(
internalComputationsClient,
systemComputation,
protocolsSetupConfig.liquidLegionsV2,
blobStorageBucket
)
}
Computation.MpcProtocolConfig.ProtocolCase.REACH_ONLY_LIQUID_LEGIONS_V2 ->
ReachOnlyLiquidLegionsV2Starter.createComputation(
internalComputationsClient,
systemComputation,
protocolsSetupConfig.reachOnlyLiquidLegionsV2,
blobStorageBucket
)
else -> error("Unknown or unsupported protocol for creation.")
}
logger.info("[id=$globalId]: Created Computation")
} catch (e: StatusException) {
Expand Down Expand Up @@ -302,6 +306,13 @@ class Herald(
systemComputation,
protocolsSetupConfig.liquidLegionsV2.externalAggregatorDuchyId
)
ComputationDetails.ProtocolCase.REACH_ONLY_LIQUID_LEGIONS_V2 ->
ReachOnlyLiquidLegionsV2Starter.updateRequisitionsAndKeySets(
token,
internalComputationsClient,
systemComputation,
protocolsSetupConfig.reachOnlyLiquidLegionsV2.externalAggregatorDuchyId
)
else -> error("Unknown or unsupported protocol.")
}
logger.info("[id=$globalId]: Confirmed Computation")
Expand All @@ -317,6 +328,8 @@ class Herald(
when (token.computationDetails.protocolCase) {
ComputationDetails.ProtocolCase.LIQUID_LEGIONS_V2 ->
LiquidLegionsV2Starter.startComputation(token, internalComputationsClient)
ComputationDetails.ProtocolCase.REACH_ONLY_LIQUID_LEGIONS_V2 ->
ReachOnlyLiquidLegionsV2Starter.startComputation(token, internalComputationsClient)
else -> error("Unknown or unsupported protocol.")
}
logger.info("[id=$globalId]: Started Computation")
Expand Down Expand Up @@ -365,8 +378,10 @@ class Herald(
} ?: return

if (
token.computationDetails.hasLiquidLegionsV2() &&
token.computationStage == LiquidLegionsV2Starter.TERMINAL_STAGE
(token.computationDetails.hasLiquidLegionsV2() &&
token.computationStage == LiquidLegionsV2Starter.TERMINAL_STAGE) ||
(token.computationDetails.hasReachOnlyLiquidLegionsV2() &&
token.computationStage == ReachOnlyLiquidLegionsV2Starter.TERMINAL_STAGE)
) {
return
}
Expand All @@ -376,6 +391,8 @@ class Herald(
endingComputationStage =
when (token.computationDetails.protocolCase) {
ComputationDetails.ProtocolCase.LIQUID_LEGIONS_V2 -> LiquidLegionsV2Starter.TERMINAL_STAGE
ComputationDetails.ProtocolCase.REACH_ONLY_LIQUID_LEGIONS_V2 ->
ReachOnlyLiquidLegionsV2Starter.TERMINAL_STAGE
else -> error { "Unknown or unsupported protocol." }
}
reason = ComputationDetails.CompletedReason.FAILED
Expand Down
4 changes: 2 additions & 2 deletions src/main/kotlin/org/wfanet/measurement/duchy/daemon/herald/LiquidLegionsV2Starter.kt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -188,7 +188,7 @@ object LiquidLegionsV2Starter {
Stage.WAIT_EXECUTION_PHASE_THREE_INPUTS,
Stage.EXECUTION_PHASE_THREE,
Stage.COMPLETE -> {
logger.info(
logger.warning(
"[id=${token.globalComputationId}]: not updating," +
" stage '$stage' is after WAIT_REQUISITIONS_AND_KEY_SET"
)
Expand Down Expand Up @@ -245,7 +245,7 @@ object LiquidLegionsV2Starter {
Stage.WAIT_EXECUTION_PHASE_THREE_INPUTS,
Stage.EXECUTION_PHASE_THREE,
Stage.COMPLETE -> {
logger.info(
logger.warning(
"[id=${token.globalComputationId}]: not starting," +
" stage '$stage' is after WAIT_TO_START"
)
Expand Down
Loading