Skip to content

Commit b893a53

Browse files
zhh2001zhh2001
committed
Add P4Switch.readMulticastGroup API with PRE entity records
The per-entity-type read template established by de3e031 and reused across e29bd3e, dd152cf, and 6e43a4b carries cleanly into the packet replication engine — with one structural difference. The PRE family (multicast groups and clone sessions) is program-agnostic: a multicast group has no P4-program name, only a controller-assigned numeric id, and P4Info carries no metadata about it. The read API for multicast groups therefore takes no String name argument: sw.readMulticastGroup() returns a builder that can optionally narrow with .groupId(long) but needs no eager P4Info lookup. The readability and pipeline-bound gates still apply. MulticastGroupEntry is a three-field record in the entity package carrying the controller-assigned multicast_group_id, an ordered list of Replica slots that define the fan-out, and an opaque Bytes metadata payload that the controller can attach for its own use (the target stores it unchanged and returns it on read; the metadata field was added in P4Runtime 1.4.0 and surfaces as empty bytes on older devices). Replica is a shared sub-record used by MulticastGroupEntry today and by the forthcoming CloneSessionEntry next. It carries the egress port, the per-clone instance id, and an ordered list of BackupReplica fallback ports. The port field is a nullable Bytes value: the P4Runtime port_kind oneof has two variants — the current bytes port field and a deprecated egress_port int32 — and jp4 surfaces both as null in this record. The two-variant treatment matches the established idiom for the action-profile-group watch_kind oneof, where WeightedMember.watchPort uses the same flat-nullable pattern. Controllers needing the deprecated egress_port int32 path can parse the wire Replica proto directly through the generated class. BackupReplica is a brand-new entity record surfaced for the first time in jp4 as a v1.5 release artefact: the corresponding P4Runtime proto message was added in spec version 1.5.0. The record is small — port bytes and an instance id, matching the proto field by field. Devices running older spec versions return empty backup_replicas lists, so existing controllers see the change as a non-breaking surface expansion. MulticastGroupReadQuery mirrors the six prior read-query interfaces in shape: a groupId(long) server-side filter, a non-default where(Predicate) client-side filter, and the five terminals (all, one, stream, allAsync, oneAsync). MulticastGroupReadQueryImpl is a private inner class on P4Switch placed adjacent to the action-profile read implementations. It reuses the same async dispatch path through outboundExecutor, the generic awaitRead and mapReadFailure helpers, and the readabilityException gate verbatim — the only new code is the entity-specific buildReadRequest (wrapping the MulticastGroupEntry in a PacketReplicationEngineEntry oneof on tag 9 of the Entity message), extractInto, flatten, and the parseMulticastGroupEntry helper that walks the Replica list and the per-Replica backup_replicas list. Six unit tests cover the new surface using the same in-process gRPC fake pattern P4SwitchReadActionProfileGroupTest established. The happy-path test exercises all three port_kind shapes in a single group (port bytes set, oneof unset, deprecated egress_port int32 set) and verifies the first surfaces a non-null port while the other two surface null. A second test covers the backup_replicas wiring with two fallback ports on a single replica. A third test covers the opaque metadata round-trip. The remaining three tests cover the pipeline-bound gate, the groupId server-side filter assembly, and the client-side where filter narrowing 3 groups to 2. The test count moves from 384 to 390. SemVer-safe pure addition: one new public method on P4Switch (readMulticastGroup), three new public records in the entity package (MulticastGroupEntry, Replica, BackupReplica), and one new public interface (MulticastGroupReadQuery). No removals, no signature changes, no behaviour changes for any existing v1.0 through v1.4 caller. The readCounter, readMeter, readRegister, readActionProfileMember, and readActionProfileGroup entries from de3e031, e29bd3e, dd152cf, and 6e43a4b are entirely unchanged.
1 parent 3e5eeed commit b893a53

6 files changed

Lines changed: 800 additions & 0 deletions

File tree

‎src/main/java/io/github/zhh2001/jp4/MulticastGroupReadQuery.java‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
package io.github.zhh2001.jp4;
2+
3+
import io.github.zhh2001.jp4.entity.MulticastGroupEntry;
4+
5+
import java.util.List;
6+
import java.util.Optional;
7+
import java.util.concurrent.CompletableFuture;
8+
import java.util.function.Predicate;
9+
import java.util.stream.Stream;
10+
11+
/**
12+
* Builder for one P4Runtime {@code Read} request against multicast
13+
* groups on the device's packet replication engine, returned by
14+
* {@code P4Switch.readMulticastGroup()}. Unlike the table-driven read
15+
* APIs ({@link CounterReadQuery}, {@link MeterReadQuery},
16+
* {@link RegisterReadQuery}, {@link ActionProfileMemberReadQuery},
17+
* {@link ActionProfileGroupReadQuery}), this query takes no P4 name —
18+
* the packet replication engine is program-agnostic and multicast
19+
* groups are addressed by a controller-assigned numeric id only.
20+
*
21+
* <p>Optionally narrow the read with {@link #groupId(long)}
22+
* (server-side filter via the wire {@code multicast_group_id} field)
23+
* and/or {@link #where(Predicate)} (client-side filter applied after
24+
* fetch), then call a terminal: {@link #all()}, {@link #one()}, or
25+
* {@link #stream()}.
26+
*
27+
* <p>An empty filter set means "every multicast group programmed on
28+
* the device". Group-id filtering happens on the device;
29+
* {@code where} filtering happens on the client after the response
30+
* has been received, the same shape
31+
* {@link CounterReadQuery#where(Predicate)} and its siblings use.
32+
*
33+
* <p>Unlike {@link ReadQuery#where}, this interface's {@link #where}
34+
* method has no default body — the interface is new in 1.5 and there
35+
* is no legacy implementer to keep working through a default. The
36+
* {@code MulticastGroupReadQueryImpl} returned by
37+
* {@code P4Switch.readMulticastGroup} is the canonical implementation.
38+
*
39+
* <p>Threading model mirrors {@link ActionProfileGroupReadQuery}
40+
* exactly: terminal operations are dispatched on the switch's outbound
41+
* executor; results return to the calling thread. A
42+
* {@code MulticastGroupReadQuery} instance is a mutable builder;
43+
* confine to a single thread.
44+
*
45+
* @since 1.5.0
46+
*/
47+
public interface MulticastGroupReadQuery {
48+
49+
/**
50+
* Restricts the read to the multicast group with the given id.
51+
* Default (unset) reads every group. Setting a second value
52+
* replaces the first.
53+
*/
54+
MulticastGroupReadQuery groupId(long multicastGroupId);
55+
56+
/**
57+
* Adds a client-side predicate that narrows the result of a subsequent
58+
* terminal call. Each call appends a predicate; entries that fail any
59+
* predicate are excluded.
60+
*
61+
* @throws NullPointerException if {@code filter} is null
62+
*/
63+
MulticastGroupReadQuery where(Predicate<? super MulticastGroupEntry> filter);
64+
65+
/** Reads every matching group into a list. */
66+
List<MulticastGroupEntry> all();
67+
68+
/**
69+
* Reads at most one group. Throws {@code P4OperationException} if the
70+
* query matches more than one — intended for fully-qualified group-id
71+
* reads where the result is 0 or 1.
72+
*/
73+
Optional<MulticastGroupEntry> one();
74+
75+
/**
76+
* Streams matching groups; the underlying gRPC iterator closes when the
77+
* stream closes. Always use with try-with-resources.
78+
*/
79+
Stream<MulticastGroupEntry> stream();
80+
81+
CompletableFuture<List<MulticastGroupEntry>> allAsync();
82+
CompletableFuture<Optional<MulticastGroupEntry>> oneAsync();
83+
}

‎src/main/java/io/github/zhh2001/jp4/P4Switch.java‎

Lines changed: 284 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,18 +3,21 @@
33
import io.github.zhh2001.jp4.entity.ActionInstance;
44
import io.github.zhh2001.jp4.entity.ActionProfileGroup;
55
import io.github.zhh2001.jp4.entity.ActionProfileMember;
6+
import io.github.zhh2001.jp4.entity.BackupReplica;
67
import io.github.zhh2001.jp4.entity.CounterData;
78
import io.github.zhh2001.jp4.entity.CounterEntry;
89
import io.github.zhh2001.jp4.entity.DigestConfig;
910
import io.github.zhh2001.jp4.entity.DigestEvent;
1011
import io.github.zhh2001.jp4.entity.DropEvent;
1112
import io.github.zhh2001.jp4.entity.IdleTimeoutEvent;
1213
import io.github.zhh2001.jp4.entity.MeterConfig;
14+
import io.github.zhh2001.jp4.entity.MulticastGroupEntry;
1315
import io.github.zhh2001.jp4.entity.MeterCounterData;
1416
import io.github.zhh2001.jp4.entity.MeterEntry;
1517
import io.github.zhh2001.jp4.entity.PacketIn;
1618
import io.github.zhh2001.jp4.entity.PacketOut;
1719
import io.github.zhh2001.jp4.entity.RegisterEntry;
20+
import io.github.zhh2001.jp4.entity.Replica;
1821
import io.github.zhh2001.jp4.entity.TableEntry;
1922
import io.github.zhh2001.jp4.entity.UpdateFailure;
2023
import io.github.zhh2001.jp4.entity.WeightedMember;
@@ -927,6 +930,39 @@ public ActionProfileGroupReadQuery readActionProfileGroup(String actionProfileNa
927930
return new ActionProfileGroupReadQueryImpl(actionProfileName, pipe);
928931
}
929932

933+
/**
934+
* Returns a builder for a P4Runtime {@code Read} request against the
935+
* multicast groups programmed on the device's packet replication engine.
936+
* Call a terminal on the returned query to dispatch the read.
937+
* Optionally narrow with {@link MulticastGroupReadQuery#groupId(long)}
938+
* (server-side filter via the wire {@code multicast_group_id} field) or
939+
* {@link MulticastGroupReadQuery#where(java.util.function.Predicate)}
940+
* (client-side filter applied to results after fetch).
941+
*
942+
* <p>Unlike the table-driven read APIs ({@link #readCounter(String)},
943+
* {@link #readMeter(String)}, {@link #readRegister(String)},
944+
* {@link #readActionProfileMember(String)},
945+
* {@link #readActionProfileGroup(String)}), this method takes no P4
946+
* name — the packet replication engine is program-agnostic and
947+
* multicast groups are addressed by controller-assigned numeric id
948+
* only. The pipeline-bound gate still applies so the call site
949+
* fails loudly if the controller has not yet bound a P4 pipeline.
950+
*
951+
* @throws P4ConnectionException if the switch is closed or the stream is broken
952+
* @throws P4PipelineException if no pipeline is bound
953+
* @since 1.5.0
954+
*/
955+
public MulticastGroupReadQuery readMulticastGroup() {
956+
P4ConnectionException gate = readabilityException();
957+
if (gate != null) throw gate;
958+
Pipeline pipe = pipeline.get();
959+
if (pipe == null) {
960+
throw new P4PipelineException(
961+
"no pipeline bound; call bindPipeline() or loadPipeline() first");
962+
}
963+
return new MulticastGroupReadQueryImpl(pipe);
964+
}
965+
930966
/**
931967
* Registers a single packet-in handler. Last-write-wins: calling this method
932968
* again replaces the prior handler. The callback runs on the same single-threaded
@@ -3532,4 +3568,252 @@ private ActionInstance parseAction(p4.v1.P4RuntimeOuterClass.Action proto) {
35323568
}
35333569
return ActionInstance.of(actionInfo.name(), params);
35343570
}
3571+
3572+
/**
3573+
* Multicast-group-read counterpart of the per-entity ReadQueryImpl
3574+
* family. Holds the Pipeline snapshot and an optional group-id
3575+
* filter, builds a {@code ReadRequest} that targets the device's
3576+
* packet replication engine (optionally narrowed to one group),
3577+
* dispatches it through the outbound executor, and parses each
3578+
* response Entity into a {@link MulticastGroupEntry} with its
3579+
* nested {@link Replica} list and per-replica
3580+
* {@link BackupReplica} entries.
3581+
*/
3582+
private final class MulticastGroupReadQueryImpl implements MulticastGroupReadQuery {
3583+
3584+
private final Pipeline pipe;
3585+
private final List<Predicate<? super MulticastGroupEntry>> filters = new ArrayList<>();
3586+
private Long groupIdFilter;
3587+
3588+
MulticastGroupReadQueryImpl(Pipeline pipe) {
3589+
this.pipe = pipe;
3590+
}
3591+
3592+
@Override
3593+
public MulticastGroupReadQuery groupId(long multicastGroupId) {
3594+
this.groupIdFilter = multicastGroupId;
3595+
return this;
3596+
}
3597+
3598+
@Override
3599+
public MulticastGroupReadQuery where(Predicate<? super MulticastGroupEntry> filter) {
3600+
Objects.requireNonNull(filter, "filter");
3601+
filters.add(filter);
3602+
return this;
3603+
}
3604+
3605+
private boolean accept(MulticastGroupEntry e) {
3606+
for (Predicate<? super MulticastGroupEntry> p : filters) {
3607+
if (!p.test(e)) return false;
3608+
}
3609+
return true;
3610+
}
3611+
3612+
@Override
3613+
public List<MulticastGroupEntry> all() {
3614+
return awaitRead(allAsync());
3615+
}
3616+
3617+
@Override
3618+
public Optional<MulticastGroupEntry> one() {
3619+
return collapseToOne(all());
3620+
}
3621+
3622+
@Override
3623+
public CompletableFuture<List<MulticastGroupEntry>> allAsync() {
3624+
P4ConnectionException gate = readabilityException();
3625+
if (gate != null) {
3626+
CompletableFuture<List<MulticastGroupEntry>> f = new CompletableFuture<>();
3627+
f.completeExceptionally(gate);
3628+
return f;
3629+
}
3630+
StreamSession sess = session.get();
3631+
if (sess == null) {
3632+
CompletableFuture<List<MulticastGroupEntry>> f = new CompletableFuture<>();
3633+
f.completeExceptionally(new P4ConnectionException("no active session"));
3634+
return f;
3635+
}
3636+
3637+
p4.v1.P4RuntimeOuterClass.ReadRequest req = buildReadRequest();
3638+
CompletableFuture<List<MulticastGroupEntry>> result = new CompletableFuture<>();
3639+
try {
3640+
outboundExecutor.execute(() -> {
3641+
try {
3642+
Iterator<p4.v1.P4RuntimeOuterClass.ReadResponse> it =
3643+
P4RuntimeGrpc.newBlockingStub(sess.channel)
3644+
.withDeadlineAfter(30, TimeUnit.SECONDS)
3645+
.read(req);
3646+
List<MulticastGroupEntry> entries = new ArrayList<>();
3647+
while (it.hasNext()) {
3648+
extractInto(it.next(), entries);
3649+
}
3650+
List<MulticastGroupEntry> filtered;
3651+
if (filters.isEmpty()) {
3652+
filtered = entries;
3653+
} else {
3654+
filtered = new ArrayList<>(entries.size());
3655+
for (MulticastGroupEntry e : entries) {
3656+
if (accept(e)) filtered.add(e);
3657+
}
3658+
}
3659+
result.complete(filtered);
3660+
} catch (StatusRuntimeException sre) {
3661+
result.completeExceptionally(mapReadFailure(sre));
3662+
} catch (RuntimeException re) {
3663+
result.completeExceptionally(re);
3664+
}
3665+
});
3666+
} catch (RejectedExecutionException ree) {
3667+
result.completeExceptionally(new P4ConnectionException("switch is closed", ree));
3668+
}
3669+
return result;
3670+
}
3671+
3672+
@Override
3673+
public CompletableFuture<Optional<MulticastGroupEntry>> oneAsync() {
3674+
return allAsync().thenApply(this::collapseToOne);
3675+
}
3676+
3677+
@Override
3678+
public Stream<MulticastGroupEntry> stream() {
3679+
P4ConnectionException gate = readabilityException();
3680+
if (gate != null) throw gate;
3681+
StreamSession sess = session.get();
3682+
if (sess == null) throw new P4ConnectionException("no active session");
3683+
p4.v1.P4RuntimeOuterClass.ReadRequest req = buildReadRequest();
3684+
3685+
Context.CancellableContext ctx = Context.current().withCancellation();
3686+
CompletableFuture<Iterator<p4.v1.P4RuntimeOuterClass.ReadResponse>> startFuture =
3687+
new CompletableFuture<>();
3688+
try {
3689+
outboundExecutor.execute(() -> {
3690+
try {
3691+
Iterator<p4.v1.P4RuntimeOuterClass.ReadResponse> it = ctx.call(() ->
3692+
P4RuntimeGrpc.newBlockingStub(sess.channel).read(req));
3693+
startFuture.complete(it);
3694+
} catch (Exception e) {
3695+
startFuture.completeExceptionally(e);
3696+
}
3697+
});
3698+
} catch (RejectedExecutionException ree) {
3699+
ctx.cancel(null);
3700+
throw new P4ConnectionException("switch is closed", ree);
3701+
}
3702+
3703+
Iterator<p4.v1.P4RuntimeOuterClass.ReadResponse> respIter;
3704+
try {
3705+
respIter = startFuture.get(30, TimeUnit.SECONDS);
3706+
} catch (TimeoutException te) {
3707+
ctx.cancel(null);
3708+
throw new P4ConnectionException(
3709+
"multicast-group read RPC timed out before stream start", te);
3710+
} catch (InterruptedException ie) {
3711+
Thread.currentThread().interrupt();
3712+
ctx.cancel(null);
3713+
throw new P4ConnectionException(
3714+
"interrupted while starting multicast-group read stream", ie);
3715+
} catch (ExecutionException ee) {
3716+
ctx.cancel(null);
3717+
Throwable cause = ee.getCause();
3718+
if (cause instanceof RuntimeException re) throw re;
3719+
throw new P4ConnectionException(
3720+
"multicast-group read RPC failed to start", cause);
3721+
}
3722+
3723+
Iterator<MulticastGroupEntry> entryIter = flatten(respIter);
3724+
Stream<MulticastGroupEntry> base = StreamSupport.stream(
3725+
Spliterators.spliteratorUnknownSize(entryIter, Spliterator.ORDERED),
3726+
/* parallel */ false
3727+
).onClose(() -> ctx.cancel(null));
3728+
return filters.isEmpty() ? base : base.filter(this::accept);
3729+
}
3730+
3731+
// ---------- helpers ------------------------------------------------
3732+
3733+
private p4.v1.P4RuntimeOuterClass.ReadRequest buildReadRequest() {
3734+
var mgBuilder = p4.v1.P4RuntimeOuterClass.MulticastGroupEntry.newBuilder();
3735+
if (groupIdFilter != null) {
3736+
mgBuilder.setMulticastGroupId(groupIdFilter.intValue());
3737+
}
3738+
var preBuilder = p4.v1.P4RuntimeOuterClass.PacketReplicationEngineEntry.newBuilder()
3739+
.setMulticastGroupEntry(mgBuilder.build());
3740+
var entity = p4.v1.P4RuntimeOuterClass.Entity.newBuilder()
3741+
.setPacketReplicationEngineEntry(preBuilder.build())
3742+
.build();
3743+
return p4.v1.P4RuntimeOuterClass.ReadRequest.newBuilder()
3744+
.setDeviceId(deviceId)
3745+
.addEntities(entity)
3746+
.build();
3747+
}
3748+
3749+
private void extractInto(p4.v1.P4RuntimeOuterClass.ReadResponse resp,
3750+
List<MulticastGroupEntry> sink) {
3751+
for (p4.v1.P4RuntimeOuterClass.Entity ent : resp.getEntitiesList()) {
3752+
if (!ent.hasPacketReplicationEngineEntry()) continue;
3753+
var pre = ent.getPacketReplicationEngineEntry();
3754+
if (!pre.hasMulticastGroupEntry()) continue;
3755+
sink.add(parseMulticastGroupEntry(pre.getMulticastGroupEntry()));
3756+
}
3757+
}
3758+
3759+
private Iterator<MulticastGroupEntry> flatten(
3760+
Iterator<p4.v1.P4RuntimeOuterClass.ReadResponse> respIter) {
3761+
return new Iterator<>() {
3762+
private Iterator<MulticastGroupEntry> currentBatch = Collections.emptyIterator();
3763+
3764+
@Override
3765+
public boolean hasNext() {
3766+
try {
3767+
while (!currentBatch.hasNext() && respIter.hasNext()) {
3768+
List<MulticastGroupEntry> batch = new ArrayList<>();
3769+
extractInto(respIter.next(), batch);
3770+
currentBatch = batch.iterator();
3771+
}
3772+
return currentBatch.hasNext();
3773+
} catch (StatusRuntimeException sre) {
3774+
throw mapReadFailure(sre);
3775+
}
3776+
}
3777+
3778+
@Override
3779+
public MulticastGroupEntry next() {
3780+
if (!hasNext()) throw new NoSuchElementException();
3781+
return currentBatch.next();
3782+
}
3783+
};
3784+
}
3785+
3786+
private MulticastGroupEntry parseMulticastGroupEntry(
3787+
p4.v1.P4RuntimeOuterClass.MulticastGroupEntry proto) {
3788+
List<Replica> replicas = new ArrayList<>(proto.getReplicasCount());
3789+
for (var r : proto.getReplicasList()) {
3790+
replicas.add(parseReplica(r));
3791+
}
3792+
Bytes metadata = Bytes.of(proto.getMetadata().toByteArray());
3793+
return new MulticastGroupEntry(proto.getMulticastGroupId(), replicas, metadata);
3794+
}
3795+
3796+
private Replica parseReplica(p4.v1.P4RuntimeOuterClass.Replica proto) {
3797+
Bytes port = proto.hasPort()
3798+
? Bytes.of(proto.getPort().toByteArray())
3799+
: null;
3800+
List<BackupReplica> backups = new ArrayList<>(proto.getBackupReplicasCount());
3801+
for (var b : proto.getBackupReplicasList()) {
3802+
backups.add(new BackupReplica(
3803+
Bytes.of(b.getPort().toByteArray()),
3804+
b.getInstance()));
3805+
}
3806+
return new Replica(port, proto.getInstance(), backups);
3807+
}
3808+
3809+
private Optional<MulticastGroupEntry> collapseToOne(List<MulticastGroupEntry> all) {
3810+
if (all.isEmpty()) return Optional.empty();
3811+
if (all.size() == 1) return Optional.of(all.get(0));
3812+
throw new P4OperationException(
3813+
"expected at most one multicast group entry, got " + all.size(),
3814+
OperationType.READ,
3815+
ErrorCode.UNKNOWN,
3816+
List.of());
3817+
}
3818+
}
35353819
}

0 commit comments

Comments
 (0)