<commit_before>/*
 * Copyright 2017-present Open Networking Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.onosproject.p4runtime.ctl;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.protobuf.lite.ProtoLiteUtils;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.StreamObserver;
import org.onlab.osgi.DefaultServiceDirectory;
import org.onlab.util.Tools;
import org.onosproject.grpc.ctl.AbstractGrpcClient;
import org.onosproject.net.pi.model.PiActionProfileId;
import org.onosproject.net.pi.model.PiCounterId;
import org.onosproject.net.pi.model.PiMeterId;
import org.onosproject.net.pi.model.PiPipeconf;
import org.onosproject.net.pi.model.PiTableId;
import org.onosproject.net.pi.runtime.PiActionGroup;
import org.onosproject.net.pi.runtime.PiActionGroupMember;
import org.onosproject.net.pi.runtime.PiActionGroupMemberId;
import org.onosproject.net.pi.runtime.PiCounterCell;
import org.onosproject.net.pi.runtime.PiCounterCellId;
import org.onosproject.net.pi.runtime.PiMeterCellConfig;
import org.onosproject.net.pi.runtime.PiMeterCellId;
import org.onosproject.net.pi.runtime.PiMulticastGroupEntry;
import org.onosproject.net.pi.runtime.PiPacketOperation;
import org.onosproject.net.pi.runtime.PiTableEntry;
import org.onosproject.net.pi.service.PiPipeconfService;
import org.onosproject.p4runtime.api.P4RuntimeClient;
import org.onosproject.p4runtime.api.P4RuntimeClientKey;
import org.onosproject.p4runtime.api.P4RuntimeEvent;
import p4.config.v1.P4InfoOuterClass.P4Info;
import p4.tmp.P4Config;
import p4.v1.P4RuntimeGrpc;
import p4.v1.P4RuntimeOuterClass;
import p4.v1.P4RuntimeOuterClass.ActionProfileGroup;
import p4.v1.P4RuntimeOuterClass.ActionProfileMember;
import p4.v1.P4RuntimeOuterClass.Entity;
import p4.v1.P4RuntimeOuterClass.ForwardingPipelineConfig;
import p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigRequest;
import p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigResponse;
import p4.v1.P4RuntimeOuterClass.MasterArbitrationUpdate;
import p4.v1.P4RuntimeOuterClass.MulticastGroupEntry;
import p4.v1.P4RuntimeOuterClass.PacketReplicationEngineEntry;
import p4.v1.P4RuntimeOuterClass.ReadRequest;
import p4.v1.P4RuntimeOuterClass.ReadResponse;
import p4.v1.P4RuntimeOuterClass.SetForwardingPipelineConfigRequest;
import p4.v1.P4RuntimeOuterClass.StreamMessageRequest;
import p4.v1.P4RuntimeOuterClass.StreamMessageResponse;
import p4.v1.P4RuntimeOuterClass.TableEntry;
import p4.v1.P4RuntimeOuterClass.Uint128;
import p4.v1.P4RuntimeOuterClass.Update;
import p4.v1.P4RuntimeOuterClass.WriteRequest;

import java.math.BigInteger;
import java.net.ConnectException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

import static com.google.common.base.Preconditions.checkNotNull;
import static java.lang.String.format;
import static java.util.Collections.singletonList;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.ACTION_PROFILE_GROUP;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.ACTION_PROFILE_MEMBER;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.PACKET_REPLICATION_ENGINE_ENTRY;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.TABLE_ENTRY;
import static p4.v1.P4RuntimeOuterClass.PacketIn;
import static p4.v1.P4RuntimeOuterClass.PacketOut;
import static p4.v1.P4RuntimeOuterClass.PacketReplicationEngineEntry.TypeCase.MULTICAST_GROUP_ENTRY;
import static p4.v1.P4RuntimeOuterClass.SetForwardingPipelineConfigRequest.Action.VERIFY_AND_COMMIT;

/**
 * Implementation of a P4Runtime client.
 */
final class P4RuntimeClientImpl extends AbstractGrpcClient implements P4RuntimeClient {

    private static final Metadata.Key<com.google.rpc.Status> STATUS_DETAILS_KEY =
            Metadata.Key.of("grpc-status-details-bin",
                            ProtoLiteUtils.metadataMarshaller(
                                    com.google.rpc.Status.getDefaultInstance()));

    private static final Map<WriteOperationType, Update.Type> UPDATE_TYPES = ImmutableMap.of(
            WriteOperationType.UNSPECIFIED, Update.Type.UNSPECIFIED,
            WriteOperationType.INSERT, Update.Type.INSERT,
            WriteOperationType.MODIFY, Update.Type.MODIFY,
            WriteOperationType.DELETE, Update.Type.DELETE
    );

    private final long p4DeviceId;
    private final P4RuntimeControllerImpl controller;
    private final P4RuntimeGrpc.P4RuntimeBlockingStub blockingStub;
    private StreamChannelManager streamChannelManager;

    // Used by this client for write requests.
    private Uint128 clientElectionId = Uint128.newBuilder().setLow(1).build();

    private final AtomicBoolean isClientMaster = new AtomicBoolean(false);

    /**
     * Default constructor.
     *
     * @param clientKey  the client key of this client
     * @param channel    gRPC channel
     * @param controller runtime client controller
     */
    P4RuntimeClientImpl(P4RuntimeClientKey clientKey, ManagedChannel channel,
                        P4RuntimeControllerImpl controller) {

        super(clientKey, channel);
        this.p4DeviceId = clientKey.p4DeviceId();
        this.controller = controller;

        //TODO Investigate use of stub deadlines instead of timeout in supplyInContext
        this.blockingStub = P4RuntimeGrpc.newBlockingStub(channel);
        this.streamChannelManager = new StreamChannelManager(channel);
    }

    @Override
    public CompletableFuture<Boolean> startStreamChannel() {
        return supplyInContext(() -> sendMasterArbitrationUpdate(false),
                               "start-initStreamChannel");
    }

    @Override
    public CompletableFuture<Boolean> becomeMaster() {
        return supplyInContext(() -> sendMasterArbitrationUpdate(true),
                               "becomeMaster");
    }

    @Override
    public boolean isMaster() {
        return streamChannelManager.isOpen() && isClientMaster.get();
    }

    @Override
    public boolean isStreamChannelOpen() {
        return streamChannelManager.isOpen();
    }

    @Override
    public CompletableFuture<Boolean> setPipelineConfig(PiPipeconf pipeconf, ByteBuffer deviceData) {
        return supplyInContext(() -> doSetPipelineConfig(pipeconf, deviceData), "setPipelineConfig");
    }

    @Override
    public boolean isPipelineConfigSet(PiPipeconf pipeconf, ByteBuffer deviceData) {
        return doIsPipelineConfigSet(pipeconf, deviceData);
    }

    @Override
    public CompletableFuture<Boolean> writeTableEntries(List<PiTableEntry> piTableEntries,
                                                        WriteOperationType opType, PiPipeconf pipeconf) {
        return supplyInContext(() -> doWriteTableEntries(piTableEntries, opType, pipeconf),
                               "writeTableEntries-" + opType.name());
    }

    @Override
    public CompletableFuture<List<PiTableEntry>> dumpTables(
            Set<PiTableId> piTableIds, boolean defaultEntries, PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpTables(piTableIds, defaultEntries, pipeconf),
                               "dumpTables-" + piTableIds.hashCode());
    }

    @Override
    public CompletableFuture<List<PiTableEntry>> dumpAllTables(PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpTables(null, false, pipeconf), "dumpAllTables");
    }

    @Override
    public CompletableFuture<Boolean> packetOut(PiPacketOperation packet, PiPipeconf pipeconf) {
        return supplyInContext(() -> doPacketOut(packet, pipeconf), "packetOut");
    }

    @Override
    public CompletableFuture<List<PiCounterCell>> readCounterCells(Set<PiCounterCellId> cellIds,
                                                                   PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadCounterCells(Lists.newArrayList(cellIds), pipeconf),
                               "readCounterCells-" + cellIds.hashCode());
    }

    @Override
    public CompletableFuture<List<PiCounterCell>> readAllCounterCells(Set<PiCounterId> counterIds,
                                                                      PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadAllCounterCells(Lists.newArrayList(counterIds), pipeconf),
                               "readAllCounterCells-" + counterIds.hashCode());
    }

    @Override
    public CompletableFuture<Boolean> writeActionGroupMembers(List<PiActionGroupMember> members,
                                                              WriteOperationType opType,
                                                              PiPipeconf pipeconf) {
        return supplyInContext(() -> doWriteActionGroupMembers(members, opType, pipeconf),
                               "writeActionGroupMembers-" + opType.name());
    }


    @Override
    public CompletableFuture<Boolean> writeActionGroup(PiActionGroup group,
                                                       WriteOperationType opType,
                                                       PiPipeconf pipeconf,
                                                       int maxMemberSize) {
        return supplyInContext(() -> doWriteActionGroup(group, opType, pipeconf, maxMemberSize),
                               "writeActionGroup-" + opType.name());
    }

    @Override
    public CompletableFuture<List<PiActionGroup>> dumpGroups(PiActionProfileId actionProfileId,
                                                             PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpGroups(actionProfileId, pipeconf),
                               "dumpGroups-" + actionProfileId.id());
    }

    @Override
    public CompletableFuture<List<PiActionGroupMemberId>> dumpActionProfileMemberIds(
            PiActionProfileId actionProfileId, PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpActionProfileMemberIds(actionProfileId, pipeconf),
                               "dumpActionProfileMemberIds-" + actionProfileId.id());
    }

    @Override
    public CompletableFuture<List<PiActionGroupMemberId>> removeActionProfileMembers(
            PiActionProfileId actionProfileId,
            List<PiActionGroupMemberId> memberIds,
            PiPipeconf pipeconf) {
        return supplyInContext(
                () -> doRemoveActionProfileMembers(actionProfileId, memberIds, pipeconf),
                "cleanupActionProfileMembers-" + actionProfileId.id());
    }

    @Override
    public CompletableFuture<Boolean> writeMeterCells(List<PiMeterCellConfig> cellIds, PiPipeconf pipeconf) {

        return supplyInContext(() -> doWriteMeterCells(cellIds, pipeconf),
                               "writeMeterCells");
    }

    @Override
    public CompletableFuture<Boolean> writePreMulticastGroupEntries(
            List<PiMulticastGroupEntry> entries,
            WriteOperationType opType) {
        return supplyInContext(() -> doWriteMulticastGroupEntries(entries, opType),
                               "writePreMulticastGroupEntries");
    }

    @Override
    public CompletableFuture<List<PiMulticastGroupEntry>> readAllMulticastGroupEntries() {
        return supplyInContext(this::doReadAllMulticastGroupEntries,
                               "readAllMulticastGroupEntries");
    }

    @Override
    public CompletableFuture<List<PiMeterCellConfig>> readMeterCells(Set<PiMeterCellId> cellIds,
                                                                     PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadMeterCells(Lists.newArrayList(cellIds), pipeconf),
                               "readMeterCells-" + cellIds.hashCode());
    }

    @Override
    public CompletableFuture<List<PiMeterCellConfig>> readAllMeterCells(Set<PiMeterId> meterIds,
                                                                        PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadAllMeterCells(Lists.newArrayList(meterIds), pipeconf),
                               "readAllMeterCells-" + meterIds.hashCode());
    }

    /* Blocking method implementations below */

    private boolean sendMasterArbitrationUpdate(boolean asMaster) {
        BigInteger newId = controller.newMasterElectionId(deviceId);
        if (asMaster) {
            // Becoming master is a race. Here we increase our chances of win
            // against other ONOS nodes in the cluster that are calling start()
            // (which is used to start the stream RPC session, not to become
            // master).
            newId = newId.add(BigInteger.valueOf(1000));
        }
        final Uint128 idMsg = bigIntegerToUint128(
                controller.newMasterElectionId(deviceId));

        log.debug("Sending arbitration update to {}... electionId={}",
                  deviceId, newId);

        streamChannelManager.send(
                StreamMessageRequest.newBuilder()
                        .setArbitration(
                                MasterArbitrationUpdate
                                        .newBuilder()
                                        .setDeviceId(p4DeviceId)
                                        .setElectionId(idMsg)
                                        .build())
                        .build());
        clientElectionId = idMsg;
        return true;
    }

    private ForwardingPipelineConfig getPipelineConfig(
            PiPipeconf pipeconf, ByteBuffer deviceData) {
        P4Info p4Info = PipeconfHelper.getP4Info(pipeconf);
        if (p4Info == null) {
            // Problem logged by PipeconfHelper.
            return null;
        }

        ForwardingPipelineConfig.Cookie pipeconfCookie = ForwardingPipelineConfig.Cookie
                .newBuilder()
                .setCookie(pipeconf.fingerprint())
                .build();

        // FIXME: This is specific to PI P4Runtime implementation.
        P4Config.P4DeviceConfig p4DeviceConfigMsg = P4Config.P4DeviceConfig
                .newBuilder()
                .setExtras(P4Config.P4DeviceConfig.Extras.getDefaultInstance())
                .setReassign(true)
                .setDeviceData(ByteString.copyFrom(deviceData))
                .build();

        return ForwardingPipelineConfig
                .newBuilder()
                .setP4Info(p4Info)
                .setP4DeviceConfig(p4DeviceConfigMsg.toByteString())
                .setCookie(pipeconfCookie)
                .build();
    }

    private boolean doIsPipelineConfigSet(PiPipeconf pipeconf, ByteBuffer deviceData) {

        GetForwardingPipelineConfigRequest request = GetForwardingPipelineConfigRequest
                .newBuilder()
                .setDeviceId(p4DeviceId)
                .setResponseType(GetForwardingPipelineConfigRequest
                                         .ResponseType.COOKIE_ONLY)
                .build();

        GetForwardingPipelineConfigResponse resp;
        try {
            resp = this.blockingStub
                    .getForwardingPipelineConfig(request);
        } catch (StatusRuntimeException ex) {
            checkGrpcException(ex);
            // FAILED_PRECONDITION means that a pipeline config was not set in
            // the first place. Don't bother logging.
            if (!ex.getStatus().getCode()
                    .equals(Status.FAILED_PRECONDITION.getCode())) {
                log.warn("Unable to get pipeline config from {}: {}",
                         deviceId, ex.getMessage());
            }
            return false;
        }
        if (!resp.getConfig().hasCookie()) {
            log.warn("{} returned GetForwardingPipelineConfigResponse " +
                             "with 'cookie' field unset",
                     deviceId);
            return false;
        }

        return resp.getConfig().getCookie().getCookie() == pipeconf.fingerprint();
    }

    private boolean doSetPipelineConfig(PiPipeconf pipeconf, ByteBuffer deviceData) {

        log.info("Setting pipeline config for {} to {}...", deviceId, pipeconf.id());

        checkNotNull(deviceData, "deviceData cannot be null");

        ForwardingPipelineConfig pipelineConfig = getPipelineConfig(pipeconf, deviceData);

        if (pipelineConfig == null) {
            // Error logged in getPipelineConfig()
            return false;<commit_msg>Reconcile the pipeline the old way if cookie is not available<commit_after>/*
 * Copyright 2017-present Open Networking Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.onosproject.p4runtime.ctl;

import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Multimap;
import com.google.common.collect.Sets;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import io.grpc.ManagedChannel;
import io.grpc.Metadata;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.grpc.protobuf.lite.ProtoLiteUtils;
import io.grpc.stub.ClientCallStreamObserver;
import io.grpc.stub.StreamObserver;
import org.onlab.osgi.DefaultServiceDirectory;
import org.onlab.util.Tools;
import org.onosproject.grpc.ctl.AbstractGrpcClient;
import org.onosproject.net.pi.model.PiActionProfileId;
import org.onosproject.net.pi.model.PiCounterId;
import org.onosproject.net.pi.model.PiMeterId;
import org.onosproject.net.pi.model.PiPipeconf;
import org.onosproject.net.pi.model.PiTableId;
import org.onosproject.net.pi.runtime.PiActionGroup;
import org.onosproject.net.pi.runtime.PiActionGroupMember;
import org.onosproject.net.pi.runtime.PiActionGroupMemberId;
import org.onosproject.net.pi.runtime.PiCounterCell;
import org.onosproject.net.pi.runtime.PiCounterCellId;
import org.onosproject.net.pi.runtime.PiMeterCellConfig;
import org.onosproject.net.pi.runtime.PiMeterCellId;
import org.onosproject.net.pi.runtime.PiMulticastGroupEntry;
import org.onosproject.net.pi.runtime.PiPacketOperation;
import org.onosproject.net.pi.runtime.PiTableEntry;
import org.onosproject.net.pi.service.PiPipeconfService;
import org.onosproject.p4runtime.api.P4RuntimeClient;
import org.onosproject.p4runtime.api.P4RuntimeClientKey;
import org.onosproject.p4runtime.api.P4RuntimeEvent;
import p4.config.v1.P4InfoOuterClass.P4Info;
import p4.tmp.P4Config;
import p4.v1.P4RuntimeGrpc;
import p4.v1.P4RuntimeOuterClass;
import p4.v1.P4RuntimeOuterClass.ActionProfileGroup;
import p4.v1.P4RuntimeOuterClass.ActionProfileMember;
import p4.v1.P4RuntimeOuterClass.Entity;
import p4.v1.P4RuntimeOuterClass.ForwardingPipelineConfig;
import p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigRequest;
import p4.v1.P4RuntimeOuterClass.GetForwardingPipelineConfigResponse;
import p4.v1.P4RuntimeOuterClass.MasterArbitrationUpdate;
import p4.v1.P4RuntimeOuterClass.MulticastGroupEntry;
import p4.v1.P4RuntimeOuterClass.PacketReplicationEngineEntry;
import p4.v1.P4RuntimeOuterClass.ReadRequest;
import p4.v1.P4RuntimeOuterClass.ReadResponse;
import p4.v1.P4RuntimeOuterClass.SetForwardingPipelineConfigRequest;
import p4.v1.P4RuntimeOuterClass.StreamMessageRequest;
import p4.v1.P4RuntimeOuterClass.StreamMessageResponse;
import p4.v1.P4RuntimeOuterClass.TableEntry;
import p4.v1.P4RuntimeOuterClass.Uint128;
import p4.v1.P4RuntimeOuterClass.Update;
import p4.v1.P4RuntimeOuterClass.WriteRequest;

import java.math.BigInteger;
import java.net.ConnectException;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;

import static com.google.common.base.Preconditions.checkNotNull;
import static java.lang.String.format;
import static java.util.Collections.singletonList;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.ACTION_PROFILE_GROUP;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.ACTION_PROFILE_MEMBER;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.PACKET_REPLICATION_ENGINE_ENTRY;
import static p4.v1.P4RuntimeOuterClass.Entity.EntityCase.TABLE_ENTRY;
import static p4.v1.P4RuntimeOuterClass.PacketIn;
import static p4.v1.P4RuntimeOuterClass.PacketOut;
import static p4.v1.P4RuntimeOuterClass.PacketReplicationEngineEntry.TypeCase.MULTICAST_GROUP_ENTRY;
import static p4.v1.P4RuntimeOuterClass.SetForwardingPipelineConfigRequest.Action.VERIFY_AND_COMMIT;

/**
 * Implementation of a P4Runtime client.
 */
final class P4RuntimeClientImpl extends AbstractGrpcClient implements P4RuntimeClient {

    private static final Metadata.Key<com.google.rpc.Status> STATUS_DETAILS_KEY =
            Metadata.Key.of("grpc-status-details-bin",
                            ProtoLiteUtils.metadataMarshaller(
                                    com.google.rpc.Status.getDefaultInstance()));

    private static final Map<WriteOperationType, Update.Type> UPDATE_TYPES = ImmutableMap.of(
            WriteOperationType.UNSPECIFIED, Update.Type.UNSPECIFIED,
            WriteOperationType.INSERT, Update.Type.INSERT,
            WriteOperationType.MODIFY, Update.Type.MODIFY,
            WriteOperationType.DELETE, Update.Type.DELETE
    );

    private final long p4DeviceId;
    private final P4RuntimeControllerImpl controller;
    private final P4RuntimeGrpc.P4RuntimeBlockingStub blockingStub;
    private StreamChannelManager streamChannelManager;

    // Used by this client for write requests.
    private Uint128 clientElectionId = Uint128.newBuilder().setLow(1).build();

    private final AtomicBoolean isClientMaster = new AtomicBoolean(false);

    /**
     * Default constructor.
     *
     * @param clientKey  the client key of this client
     * @param channel    gRPC channel
     * @param controller runtime client controller
     */
    P4RuntimeClientImpl(P4RuntimeClientKey clientKey, ManagedChannel channel,
                        P4RuntimeControllerImpl controller) {

        super(clientKey, channel);
        this.p4DeviceId = clientKey.p4DeviceId();
        this.controller = controller;

        //TODO Investigate use of stub deadlines instead of timeout in supplyInContext
        this.blockingStub = P4RuntimeGrpc.newBlockingStub(channel);
        this.streamChannelManager = new StreamChannelManager(channel);
    }

    @Override
    public CompletableFuture<Boolean> startStreamChannel() {
        return supplyInContext(() -> sendMasterArbitrationUpdate(false),
                               "start-initStreamChannel");
    }

    @Override
    public CompletableFuture<Boolean> becomeMaster() {
        return supplyInContext(() -> sendMasterArbitrationUpdate(true),
                               "becomeMaster");
    }

    @Override
    public boolean isMaster() {
        return streamChannelManager.isOpen() && isClientMaster.get();
    }

    @Override
    public boolean isStreamChannelOpen() {
        return streamChannelManager.isOpen();
    }

    @Override
    public CompletableFuture<Boolean> setPipelineConfig(PiPipeconf pipeconf, ByteBuffer deviceData) {
        return supplyInContext(() -> doSetPipelineConfig(pipeconf, deviceData), "setPipelineConfig");
    }

    @Override
    public boolean isPipelineConfigSet(PiPipeconf pipeconf, ByteBuffer deviceData) {
        return doIsPipelineConfigSet(pipeconf, deviceData);
    }

    @Override
    public CompletableFuture<Boolean> writeTableEntries(List<PiTableEntry> piTableEntries,
                                                        WriteOperationType opType, PiPipeconf pipeconf) {
        return supplyInContext(() -> doWriteTableEntries(piTableEntries, opType, pipeconf),
                               "writeTableEntries-" + opType.name());
    }

    @Override
    public CompletableFuture<List<PiTableEntry>> dumpTables(
            Set<PiTableId> piTableIds, boolean defaultEntries, PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpTables(piTableIds, defaultEntries, pipeconf),
                               "dumpTables-" + piTableIds.hashCode());
    }

    @Override
    public CompletableFuture<List<PiTableEntry>> dumpAllTables(PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpTables(null, false, pipeconf), "dumpAllTables");
    }

    @Override
    public CompletableFuture<Boolean> packetOut(PiPacketOperation packet, PiPipeconf pipeconf) {
        return supplyInContext(() -> doPacketOut(packet, pipeconf), "packetOut");
    }

    @Override
    public CompletableFuture<List<PiCounterCell>> readCounterCells(Set<PiCounterCellId> cellIds,
                                                                   PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadCounterCells(Lists.newArrayList(cellIds), pipeconf),
                               "readCounterCells-" + cellIds.hashCode());
    }

    @Override
    public CompletableFuture<List<PiCounterCell>> readAllCounterCells(Set<PiCounterId> counterIds,
                                                                      PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadAllCounterCells(Lists.newArrayList(counterIds), pipeconf),
                               "readAllCounterCells-" + counterIds.hashCode());
    }

    @Override
    public CompletableFuture<Boolean> writeActionGroupMembers(List<PiActionGroupMember> members,
                                                              WriteOperationType opType,
                                                              PiPipeconf pipeconf) {
        return supplyInContext(() -> doWriteActionGroupMembers(members, opType, pipeconf),
                               "writeActionGroupMembers-" + opType.name());
    }


    @Override
    public CompletableFuture<Boolean> writeActionGroup(PiActionGroup group,
                                                       WriteOperationType opType,
                                                       PiPipeconf pipeconf,
                                                       int maxMemberSize) {
        return supplyInContext(() -> doWriteActionGroup(group, opType, pipeconf, maxMemberSize),
                               "writeActionGroup-" + opType.name());
    }

    @Override
    public CompletableFuture<List<PiActionGroup>> dumpGroups(PiActionProfileId actionProfileId,
                                                             PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpGroups(actionProfileId, pipeconf),
                               "dumpGroups-" + actionProfileId.id());
    }

    @Override
    public CompletableFuture<List<PiActionGroupMemberId>> dumpActionProfileMemberIds(
            PiActionProfileId actionProfileId, PiPipeconf pipeconf) {
        return supplyInContext(() -> doDumpActionProfileMemberIds(actionProfileId, pipeconf),
                               "dumpActionProfileMemberIds-" + actionProfileId.id());
    }

    @Override
    public CompletableFuture<List<PiActionGroupMemberId>> removeActionProfileMembers(
            PiActionProfileId actionProfileId,
            List<PiActionGroupMemberId> memberIds,
            PiPipeconf pipeconf) {
        return supplyInContext(
                () -> doRemoveActionProfileMembers(actionProfileId, memberIds, pipeconf),
                "cleanupActionProfileMembers-" + actionProfileId.id());
    }

    @Override
    public CompletableFuture<Boolean> writeMeterCells(List<PiMeterCellConfig> cellIds, PiPipeconf pipeconf) {

        return supplyInContext(() -> doWriteMeterCells(cellIds, pipeconf),
                               "writeMeterCells");
    }

    @Override
    public CompletableFuture<Boolean> writePreMulticastGroupEntries(
            List<PiMulticastGroupEntry> entries,
            WriteOperationType opType) {
        return supplyInContext(() -> doWriteMulticastGroupEntries(entries, opType),
                               "writePreMulticastGroupEntries");
    }

    @Override
    public CompletableFuture<List<PiMulticastGroupEntry>> readAllMulticastGroupEntries() {
        return supplyInContext(this::doReadAllMulticastGroupEntries,
                               "readAllMulticastGroupEntries");
    }

    @Override
    public CompletableFuture<List<PiMeterCellConfig>> readMeterCells(Set<PiMeterCellId> cellIds,
                                                                     PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadMeterCells(Lists.newArrayList(cellIds), pipeconf),
                               "readMeterCells-" + cellIds.hashCode());
    }

    @Override
    public CompletableFuture<List<PiMeterCellConfig>> readAllMeterCells(Set<PiMeterId> meterIds,
                                                                        PiPipeconf pipeconf) {
        return supplyInContext(() -> doReadAllMeterCells(Lists.newArrayList(meterIds), pipeconf),
                               "readAllMeterCells-" + meterIds.hashCode());
    }

    /* Blocking method implementations below */

    private boolean sendMasterArbitrationUpdate(boolean asMaster) {
        BigInteger newId = controller.newMasterElectionId(deviceId);
        if (asMaster) {
            // Becoming master is a race. Here we increase our chances of win
            // against other ONOS nodes in the cluster that are calling start()
            // (which is used to start the stream RPC session, not to become
            // master).
            newId = newId.add(BigInteger.valueOf(1000));
        }
        final Uint128 idMsg = bigIntegerToUint128(
                controller.newMasterElectionId(deviceId));

        log.debug("Sending arbitration update to {}... electionId={}",
                  deviceId, newId);

        streamChannelManager.send(
                StreamMessageRequest.newBuilder()
                        .setArbitration(
                                MasterArbitrationUpdate
                                        .newBuilder()
                                        .setDeviceId(p4DeviceId)
                                        .setElectionId(idMsg)
                                        .build())
                        .build());
        clientElectionId = idMsg;
        return true;
    }

    private ForwardingPipelineConfig getPipelineConfig(
            PiPipeconf pipeconf, ByteBuffer deviceData) {
        P4Info p4Info = PipeconfHelper.getP4Info(pipeconf);
        if (p4Info == null) {
            // Problem logged by PipeconfHelper.
            return null;
        }

        ForwardingPipelineConfig.Cookie pipeconfCookie = ForwardingPipelineConfig.Cookie
                .newBuilder()
                .setCookie(pipeconf.fingerprint())
                .build();

        // FIXME: This is specific to PI P4Runtime implementation.
        P4Config.P4DeviceConfig p4DeviceConfigMsg = P4Config.P4DeviceConfig
                .newBuilder()
                .setExtras(P4Config.P4DeviceConfig.Extras.getDefaultInstance())
                .setReassign(true)
                .setDeviceData(ByteString.copyFrom(deviceData))
                .build();

        return ForwardingPipelineConfig
                .newBuilder()
                .setP4Info(p4Info)
                .setP4DeviceConfig(p4DeviceConfigMsg.toByteString())
                .setCookie(pipeconfCookie)
                .build();
    }

    private boolean doIsPipelineConfigSet(PiPipeconf pipeconf, ByteBuffer deviceData) {

        GetForwardingPipelineConfigRequest request = GetForwardingPipelineConfigRequest
                .newBuilder()
                .setDeviceId(p4DeviceId)
                .setResponseType(GetForwardingPipelineConfigRequest
                                         .ResponseType.COOKIE_ONLY)
                .build();

        GetForwardingPipelineConfigResponse resp;
        try {
            resp = this.blockingStub
                    .getForwardingPipelineConfig(request);
        } catch (StatusRuntimeException ex) {
            checkGrpcException(ex);
            // FAILED_PRECONDITION means that a pipeline config was not set in
            // the first place. Don't bother logging.
            if (!ex.getStatus().getCode()
                    .equals(Status.FAILED_PRECONDITION.getCode())) {
                log.warn("Unable to get pipeline config from {}: {}",
                         deviceId, ex.getMessage());
            }
            return false;
        }
        if (!resp.getConfig().hasCookie()) {
            log.warn("{} returned GetForwardingPipelineConfigResponse " +
                             "with 'cookie' field unset. " +
                             "Will try by comparing 'device_data'...",
                     deviceId);
            return doIsPipelineConfigSetWithData(pipeconf, deviceData);
        }

        return resp.getConfig().getCookie().getCookie() == pipeconf.fingerprint();
    }

    private boolean doIsPipelineConfigSetWithData(PiPipeconf pipeconf, ByteBuffer deviceData) {

        GetForwardingPipelineConfigRequest request = GetForwardingPipelineConfigRequest
                .newBuilder()
                .setDeviceId(p4DeviceId)
                .build();

        GetForwardingPipelineConfigResponse resp;
        try {
            resp = this.blockingStub
                    .getForwardingPipelineConfig(request);
        } catch (StatusRuntimeException ex) {
            checkGrpcException(ex);
            return false;
        }

        ForwardingPipelineConfig expectedConfig = getPipelineConfig(
                pipeconf, deviceData);

        if (expectedConfig == null) {
            return false;
        }
        if (!resp.hasConfig()) {
            log.warn("{} returned GetForwardingPipelineConfigResponse " +
                             "with 'config' field unset",
                     deviceId);
            return false;
        }
        if (resp.getConfig().getP4DeviceConfig().isEmpty()
                && !expectedConfig.getP4DeviceConfig().isEmpty()) {
            // Don't bother with a warn or error since we don't really allow
            // updating the pipeline to a different one. So the P4Info should be
            // enough for us.
            log.debug("{} returned GetForwardingPipelineConfigResponse " +
                              "with empty 'p4_device_config' field, " +
                              "equality will be based only on P4Info",
                      deviceId);
            return resp.getConfig().getP4Info().equals(
                    expectedConfig.getP4Info());
        } else {
            return resp.getConfig().getP4DeviceConfig()
                    .equals(expectedConfig.getP4DeviceConfig())
                    && resp.getConfig().getP4Info()
                    .equals(expectedConfig.getP4Info());
        }
    }

    private boolean doSetPipelineConfig(PiPipeconf pipeconf, ByteBuffer deviceData) {

        log.info("Setting pipeline config for {} to {}...", deviceId, pipeconf.id());

        checkNotNull(deviceData, "deviceData cannot be null");

        ForwardingPipelineConfig pipelineConfig = getPipelineConfig(pipeconf, deviceData);

        if (pipelineConfig == null) {
            // Error logged in getPipelineConfig()
            return false;
{'fraction_non_alphanumeric': 0.08232616118367568, 'fraction_numerical': 0.008780011923472983, 'mean_word_length': 2.5548598400924765, 'pattern_counts': {'":': 0, '<': 85, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 64, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5222078', 'n_tokens_mistral': 9788, 'n_tokens_neox': 9507, 'n_words': 1853}