Allow re-creating clients for the same P4Runtime addr-port
Change-Id: Ib3de10d047f52dd28511e71385773d4b4a9ad74f
diff --git a/drivers/p4runtime/BUCK b/drivers/p4runtime/BUCK
index 0c9f23d..ea135ef 100644
--- a/drivers/p4runtime/BUCK
+++ b/drivers/p4runtime/BUCK
@@ -5,7 +5,6 @@
'//lib:KRYO',
'//protocols/p4runtime/api:onos-protocols-p4runtime-api',
'//incubator/grpc-dependencies:grpc-core-repkg-' + GRPC_VER,
- '//lib:grpc-netty-' + GRPC_VER,
'//core/store/serializers:onos-core-serializers',
]
diff --git a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java
index 8783adc..8e2b54d 100644
--- a/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java
+++ b/drivers/p4runtime/src/main/java/org/onosproject/drivers/p4runtime/AbstractP4RuntimeHandlerBehaviour.java
@@ -16,8 +16,6 @@
package org.onosproject.drivers.p4runtime;
-import io.grpc.ManagedChannelBuilder;
-import io.grpc.netty.NettyChannelBuilder;
import org.onosproject.net.Device;
import org.onosproject.net.DeviceId;
import org.onosproject.net.device.DeviceService;
@@ -100,9 +98,9 @@
deviceId = handler().data().deviceId();
controller = handler().get(P4RuntimeController.class);
- String serverAddr = this.data().value(P4RUNTIME_SERVER_ADDR_KEY);
- String serverPortString = this.data().value(P4RUNTIME_SERVER_PORT_KEY);
- String p4DeviceIdString = this.data().value(P4RUNTIME_DEVICE_ID_KEY);
+ final String serverAddr = this.data().value(P4RUNTIME_SERVER_ADDR_KEY);
+ final String serverPortString = this.data().value(P4RUNTIME_SERVER_PORT_KEY);
+ final String p4DeviceIdString = this.data().value(P4RUNTIME_DEVICE_ID_KEY);
if (serverAddr == null || serverPortString == null || p4DeviceIdString == null) {
log.warn("Unable to create client for {}, missing driver data key (required is {}, {}, and {})",
@@ -110,11 +108,9 @@
return false;
}
- ManagedChannelBuilder channelBuilder = NettyChannelBuilder
- .forAddress(serverAddr, Integer.valueOf(serverPortString))
- .usePlaintext(true);
-
- if (!controller.createClient(deviceId, Long.parseUnsignedLong(p4DeviceIdString), channelBuilder)) {
+ if (!controller.createClient(deviceId, serverAddr,
+ Integer.parseUnsignedInt(serverPortString),
+ Long.parseUnsignedLong(p4DeviceIdString))) {
log.warn("Unable to create client for {}, aborting operation", deviceId);
return false;
}
diff --git a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java
index d8bb9c7..3cd81f2 100644
--- a/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java
+++ b/protocols/p4runtime/api/src/main/java/org/onosproject/p4runtime/api/P4RuntimeController.java
@@ -17,7 +17,6 @@
package org.onosproject.p4runtime.api;
import com.google.common.annotations.Beta;
-import io.grpc.ManagedChannelBuilder;
import org.onosproject.event.ListenerService;
import org.onosproject.net.DeviceId;
import org.onosproject.net.device.ChannelListener;
@@ -29,18 +28,30 @@
public interface P4RuntimeController extends ListenerService<P4RuntimeEvent, P4RuntimeEventListener> {
/**
- * Instantiates a new client to operate on the device identified by the given information and reachable using the
- * given gRPC channel builder. As a result of this method, a {@link P4RuntimeClient} can be later obtained by
- * invoking {@link #getClient(DeviceId)}. Only one client can exist for the same device identifier. Returns true if
- * the client was created and the channel to the device is open, false otherwise.
+ * Instantiates a new client to operate on a P4Runtime device identified by
+ * the given information. As a result of this method, a {@link
+ * P4RuntimeClient} can be later obtained by invoking {@link
+ * #getClient(DeviceId)}. Returns true if the client was created and the
+ * channel to the device is open, false otherwise.
+ * <p>
+ * Only one client can exist for the same device ID. Calls to this method
+ * are idempotent for the same [device ID, address, port, p4DeviceId] tuple,
+ * i.e. returns true if such client already exists but a new one is not
+ * created. Throws an {@link IllegalStateException} if a client for device
+ * ID already exists but for different [address, port, p4DeviceId].
*
- * @param deviceId device identifier
- * @param p4DeviceId P4Runtime-specific device identifier
- * @param channelBuilder gRPC channel builder pointing at the P4Runtime server in execution on the device
- * @return true if the client was created and the channel to the device is open
- * @throws IllegalStateException if a client already exists for the given device identifier
+ * @param deviceId device identifier
+ * @param serverAddr address of the P4Runtime server
+ * @param serverPort port of the P4Runtime server
+ * @param p4DeviceId P4Runtime-specific device identifier
+ * @return true if the client was created and the channel to the device is
+ * open
+ * @throws IllegalStateException if a client already exists for this device
+ * ID but for different [address, port,
+ * p4DeviceId].
*/
- boolean createClient(DeviceId deviceId, long p4DeviceId, ManagedChannelBuilder channelBuilder);
+ boolean createClient(DeviceId deviceId, String serverAddr, int serverPort,
+ long p4DeviceId);
/**
* Returns a client to operate on the given device.
@@ -70,7 +81,7 @@
/**
* Returns true if the P4Runtime server running on the given device is reachable, i.e. the channel is open and the
* server is able to respond to RPCs, false otherwise. Reachability can be tested only if a client was previously
- * created using {@link #createClient(DeviceId, long, ManagedChannelBuilder)}, otherwise this method returns false.
+ * created using {@link #createClient(DeviceId, String, int, long)}, otherwise this method returns false.
*
* @param deviceId device identifier.
* @return true if a client was created and is able to contact the P4Runtime server, false otherwise.
diff --git a/protocols/p4runtime/ctl/BUCK b/protocols/p4runtime/ctl/BUCK
index 90a0266..cc62acb 100644
--- a/protocols/p4runtime/ctl/BUCK
+++ b/protocols/p4runtime/ctl/BUCK
@@ -8,6 +8,7 @@
'//protocols/p4runtime/proto:onos-protocols-p4runtime-proto',
'//incubator/grpc-dependencies:grpc-core-repkg-' + GRPC_VER,
'//lib:grpc-stub-' + GRPC_VER,
+ '//lib:grpc-netty-' + GRPC_VER,
'//lib:protobuf-java-' + PROTOBUF_VER,
]
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/ClientKey.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/ClientKey.java
new file mode 100644
index 0000000..ee5e2a4
--- /dev/null
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/ClientKey.java
@@ -0,0 +1,113 @@
+/*
+ * Copyright 2018-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.base.MoreObjects;
+import org.onosproject.net.DeviceId;
+
+import java.util.Objects;
+
+/**
+ * Key the uniquely identifies a P4Runtime client.
+ */
+final class ClientKey {
+
+ private final DeviceId deviceId;
+ private final String serverAddr;
+ private final int serverPort;
+ private final long p4DeviceId;
+
+ /**
+ * Creates a new client key.
+ *
+ * @param deviceId ONOS device ID
+ * @param serverAddr P4Runtime server address
+ * @param serverPort P4Runtime server port
+ * @param p4DeviceId P4Runtime server-internal device ID
+ */
+ ClientKey(DeviceId deviceId, String serverAddr, int serverPort, long p4DeviceId) {
+ this.deviceId = deviceId;
+ this.serverAddr = serverAddr;
+ this.serverPort = serverPort;
+ this.p4DeviceId = p4DeviceId;
+ }
+
+ /**
+ * Returns the device ID.
+ *
+ * @return device ID.
+ */
+ public DeviceId deviceId() {
+ return deviceId;
+ }
+
+ /**
+ * Returns the P4Runtime server address.
+ *
+ * @return P4Runtime server address
+ */
+ public String serverAddr() {
+ return serverAddr;
+ }
+
+ /**
+ * Returns the P4Runtime server port.
+ *
+ * @return P4Runtime server port
+ */
+ public int serverPort() {
+ return serverPort;
+ }
+
+ /**
+ * Returns the P4Runtime server-internal device ID.
+ *
+ * @return P4Runtime server-internal device ID
+ */
+ public long p4DeviceId() {
+ return p4DeviceId;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(serverAddr, serverPort, p4DeviceId);
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final ClientKey other = (ClientKey) obj;
+ return Objects.equals(this.serverAddr, other.serverAddr)
+ && Objects.equals(this.serverPort, other.serverPort)
+ && Objects.equals(this.p4DeviceId, other.p4DeviceId);
+ }
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("deviceId", deviceId)
+ .add("serverAddr", serverAddr)
+ .add("serverPort", serverPort)
+ .add("p4DeviceId", p4DeviceId)
+ .toString();
+ }
+}
diff --git a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/P4RuntimeControllerImpl.java b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/P4RuntimeControllerImpl.java
index 47ebe68..987356b 100644
--- a/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/P4RuntimeControllerImpl.java
+++ b/protocols/p4runtime/ctl/src/main/java/org/onosproject/p4runtime/ctl/P4RuntimeControllerImpl.java
@@ -24,6 +24,7 @@
import io.grpc.ManagedChannelBuilder;
import io.grpc.NameResolverProvider;
import io.grpc.internal.DnsNameResolverProvider;
+import io.grpc.netty.NettyChannelBuilder;
import org.apache.felix.scr.annotations.Activate;
import org.apache.felix.scr.annotations.Component;
import org.apache.felix.scr.annotations.Deactivate;
@@ -53,7 +54,6 @@
import java.util.concurrent.locks.ReentrantReadWriteLock;
import static com.google.common.base.Preconditions.checkNotNull;
-import static java.lang.String.format;
import static org.slf4j.LoggerFactory.getLogger;
/**
@@ -69,7 +69,8 @@
private static final int DEVICE_LOCK_EXPIRE_TIME_IN_MIN = 10;
private final Logger log = getLogger(getClass());
private final NameResolverProvider nameResolverProvider = new DnsNameResolverProvider();
- private final Map<DeviceId, P4RuntimeClient> clients = Maps.newHashMap();
+ private final Map<DeviceId, ClientKey> deviceIdToClientKey = Maps.newHashMap();
+ private final Map<ClientKey, P4RuntimeClient> clientKeyToClient = Maps.newHashMap();
private final Map<DeviceId, GrpcChannelId> channelIds = Maps.newHashMap();
private final Map<DeviceId, List<ChannelListener>> channelListeners = Maps.newConcurrentMap();
private final LoadingCache<DeviceId, ReadWriteLock> deviceLocks = CacheBuilder.newBuilder()
@@ -84,10 +85,10 @@
private AtomicCounter electionIdGenerator;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- public GrpcController grpcController;
+ private GrpcController grpcController;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
- public StorageService storageService;
+ private StorageService storageService;
@Activate
public void activate() {
@@ -105,31 +106,44 @@
log.info("Stopped");
}
-
@Override
- public boolean createClient(DeviceId deviceId, long p4DeviceId, ManagedChannelBuilder channelBuilder) {
+ public boolean createClient(DeviceId deviceId, String serverAddr,
+ int serverPort, long p4DeviceId) {
checkNotNull(deviceId);
- checkNotNull(channelBuilder);
+ checkNotNull(serverAddr);
+
+ ClientKey newKey = new ClientKey(deviceId, serverAddr, serverPort, p4DeviceId);
+
+ ManagedChannelBuilder channelBuilder = NettyChannelBuilder
+ .forAddress(serverAddr, serverPort)
+ .usePlaintext(true);
deviceLocks.getUnchecked(deviceId).writeLock().lock();
- log.info("Creating client for {} (with internal device id {})...", deviceId, p4DeviceId);
+ log.info("Creating client for {} (server={}:{}, p4DeviceId={})...",
+ deviceId, serverAddr, serverPort, p4DeviceId);
try {
- if (clients.containsKey(deviceId)) {
- // TODO might want to consider a more fine-grained check such as same port/p4DeviceId
- log.warn("A client already exists for {}", deviceId);
- throw new IllegalStateException(format("A client already exists for %s", deviceId));
+ if (deviceIdToClientKey.containsKey(deviceId)) {
+ final ClientKey existingKey = deviceIdToClientKey.get(deviceId);
+ if (newKey.equals(existingKey)) {
+ return true;
+ } else {
+ throw new IllegalStateException(
+ "A client for the same device ID but different " +
+ "server endpoints already exists");
+ }
} else {
- return doCreateClient(deviceId, p4DeviceId, channelBuilder);
+ return doCreateClient(newKey, channelBuilder);
}
} finally {
deviceLocks.getUnchecked(deviceId).writeLock().unlock();
}
}
- private boolean doCreateClient(DeviceId deviceId, long p4DeviceId, ManagedChannelBuilder channelBuilder) {
+ private boolean doCreateClient(ClientKey clientKey, ManagedChannelBuilder channelBuilder) {
- GrpcChannelId channelId = GrpcChannelId.of(deviceId, "p4runtime");
+ GrpcChannelId channelId = GrpcChannelId.of(clientKey.deviceId(),
+ "p4runtime-" + clientKey.p4DeviceId());
// Channel defaults.
channelBuilder.nameResolverFactory(nameResolverProvider);
@@ -138,14 +152,17 @@
try {
channel = grpcController.connectChannel(channelId, channelBuilder);
} catch (IOException e) {
- log.warn("Unable to connect to gRPC server of {}: {}", deviceId, e.getMessage());
+ log.warn("Unable to connect to gRPC server of {}: {}",
+ clientKey.deviceId(), e.getMessage());
return false;
}
- P4RuntimeClient client = new P4RuntimeClientImpl(deviceId, p4DeviceId, channel, this);
+ P4RuntimeClient client = new P4RuntimeClientImpl(
+ clientKey.deviceId(), clientKey.p4DeviceId(), channel, this);
- channelIds.put(deviceId, channelId);
- clients.put(deviceId, client);
+ channelIds.put(clientKey.deviceId(), channelId);
+ deviceIdToClientKey.put(clientKey.deviceId(), clientKey);
+ clientKeyToClient.put(clientKey, client);
return true;
}
@@ -156,7 +173,11 @@
deviceLocks.getUnchecked(deviceId).readLock().lock();
try {
- return clients.get(deviceId);
+ if (!deviceIdToClientKey.containsKey(deviceId)) {
+ return null;
+ } else {
+ return clientKeyToClient.get(deviceIdToClientKey.get(deviceId));
+ }
} finally {
deviceLocks.getUnchecked(deviceId).readLock().unlock();
}
@@ -168,10 +189,11 @@
deviceLocks.getUnchecked(deviceId).writeLock().lock();
try {
- if (clients.containsKey(deviceId)) {
- clients.get(deviceId).shutdown();
+ if (deviceIdToClientKey.containsKey(deviceId)) {
+ final ClientKey clientKey = deviceIdToClientKey.get(deviceId);
grpcController.disconnectChannel(channelIds.get(deviceId));
- clients.remove(deviceId);
+ clientKeyToClient.remove(clientKey).shutdown();
+ deviceIdToClientKey.remove(deviceId);
channelIds.remove(deviceId);
}
} finally {
@@ -184,7 +206,7 @@
deviceLocks.getUnchecked(deviceId).readLock().lock();
try {
- return clients.containsKey(deviceId);
+ return deviceIdToClientKey.containsKey(deviceId);
} finally {
deviceLocks.getUnchecked(deviceId).readLock().unlock();
}
@@ -196,8 +218,8 @@
deviceLocks.getUnchecked(deviceId).readLock().lock();
try {
- if (!clients.containsKey(deviceId)) {
- log.warn("No client for {}, can't check for reachability", deviceId);
+ if (!deviceIdToClientKey.containsKey(deviceId)) {
+ log.debug("No client for {}, can't check for reachability", deviceId);
return false;
}
@@ -239,7 +261,7 @@
});
}
- public void postEvent(P4RuntimeEvent event) {
+ void postEvent(P4RuntimeEvent event) {
if (event.type().equals(P4RuntimeEvent.Type.CHANNEL_EVENT)) {
DefaultChannelEvent channelError = (DefaultChannelEvent) event.subject();
DeviceId deviceId = event.subject().deviceId();
@@ -270,4 +292,5 @@
post(event);
}
}
+
}
diff --git a/tools/dev/mininet/bmv2.py b/tools/dev/mininet/bmv2.py
index b6ac2f6..c4ec15f 100644
--- a/tools/dev/mininet/bmv2.py
+++ b/tools/dev/mininet/bmv2.py
@@ -1,15 +1,14 @@
+import json
import multiprocessing
import os
-import socket
+import random
import re
-import json
+import socket
import threading
-import urllib2
-
import time
+import urllib2
from contextlib import closing
-
-from mininet.log import info, warn, error
+from mininet.log import info, warn
from mininet.node import Switch, Host
SIMPLE_SWITCH_GRPC = 'simple_switch_grpc'
@@ -322,7 +321,7 @@
except AttributeError:
clist = controllers
assert len(clist) > 0
- return clist[0].IP()
+ return random.choice(clist).IP()
def killBmv2(self, log=False):
if self.bmv2popen is not None: