Committed by
Gerrit Code Review
ONOS-4278 Implemented BMv2 control plane server and packet-out support
Change-Id: I4d9027b232dea31d1091c980fb040ec93da9473d
Showing
6 changed files
with
472 additions
and
27 deletions
... | @@ -34,11 +34,11 @@ | ... | @@ -34,11 +34,11 @@ |
34 | 34 | ||
35 | <properties> | 35 | <properties> |
36 | <!-- BMv2 Commit ID and Thrift version --> | 36 | <!-- BMv2 Commit ID and Thrift version --> |
37 | - <bmv2.commit>a012ee4124c1892a91a359660824d311d5d7fe88</bmv2.commit> | 37 | + <bmv2.commit>4421bafd6d26740b0bbf802c2e9f9f54c1211b13</bmv2.commit> |
38 | <bmv2.thrift.version>0.9.3</bmv2.thrift.version> | 38 | <bmv2.thrift.version>0.9.3</bmv2.thrift.version> |
39 | <!-- Do not change below --> | 39 | <!-- Do not change below --> |
40 | <bmv2.baseurl> | 40 | <bmv2.baseurl> |
41 | - https://raw.githubusercontent.com/p4lang/behavioral-model/${bmv2.commit} | 41 | + https://raw.githubusercontent.com/ccascone/behavioral-model/${bmv2.commit} |
42 | </bmv2.baseurl> | 42 | </bmv2.baseurl> |
43 | <bmv2.thrift.srcdir>${project.basedir}/src/main/thrift</bmv2.thrift.srcdir> | 43 | <bmv2.thrift.srcdir>${project.basedir}/src/main/thrift</bmv2.thrift.srcdir> |
44 | <thrift.path>${project.build.directory}/thrift-compiler/</thrift.path> | 44 | <thrift.path>${project.build.directory}/thrift-compiler/</thrift.path> |
... | @@ -56,6 +56,10 @@ | ... | @@ -56,6 +56,10 @@ |
56 | <artifactId>onos-api</artifactId> | 56 | <artifactId>onos-api</artifactId> |
57 | <version>${project.version}</version> | 57 | <version>${project.version}</version> |
58 | </dependency> | 58 | </dependency> |
59 | + <dependency> | ||
60 | + <groupId>org.apache.felix</groupId> | ||
61 | + <artifactId>org.apache.felix.scr.annotations</artifactId> | ||
62 | + </dependency> | ||
59 | </dependencies> | 63 | </dependencies> |
60 | 64 | ||
61 | <repositories> | 65 | <repositories> |
... | @@ -84,7 +88,7 @@ | ... | @@ -84,7 +88,7 @@ |
84 | <executions> | 88 | <executions> |
85 | <execution> | 89 | <execution> |
86 | <id>download-bmv2-thrift-standard</id> | 90 | <id>download-bmv2-thrift-standard</id> |
87 | - <phase>validate</phase> | 91 | + <phase>initialize</phase> |
88 | <goals> | 92 | <goals> |
89 | <goal>download-single</goal> | 93 | <goal>download-single</goal> |
90 | </goals> | 94 | </goals> |
... | @@ -133,6 +137,20 @@ | ... | @@ -133,6 +137,20 @@ |
133 | <toDir>${bmv2.thrift.srcdir}</toDir> | 137 | <toDir>${bmv2.thrift.srcdir}</toDir> |
134 | </configuration> | 138 | </configuration> |
135 | </execution> | 139 | </execution> |
140 | + <execution> | ||
141 | + <id>download-bmv2-thrift-simple_switch-cpservice</id> | ||
142 | + <phase>initialize</phase> | ||
143 | + <goals> | ||
144 | + <goal>download-single</goal> | ||
145 | + </goals> | ||
146 | + <configuration> | ||
147 | + <url>${bmv2.baseurl}</url> | ||
148 | + <fromFile> | ||
149 | + targets/simple_switch/thrift/control_plane.thrift | ||
150 | + </fromFile> | ||
151 | + <toDir>${bmv2.thrift.srcdir}</toDir> | ||
152 | + </configuration> | ||
153 | + </execution> | ||
136 | </executions> | 154 | </executions> |
137 | </plugin> | 155 | </plugin> |
138 | <!-- Extract Thrift compiler --> | 156 | <!-- Extract Thrift compiler --> |
... | @@ -201,11 +219,12 @@ | ... | @@ -201,11 +219,12 @@ |
201 | <version>0.1.11</version> | 219 | <version>0.1.11</version> |
202 | <configuration> | 220 | <configuration> |
203 | <thriftExecutable>${thrift.path}/${thrift.filename}</thriftExecutable> | 221 | <thriftExecutable>${thrift.path}/${thrift.filename}</thriftExecutable> |
222 | + <outputDirectory>${project.build.directory}/generated-sources</outputDirectory> | ||
204 | </configuration> | 223 | </configuration> |
205 | <executions> | 224 | <executions> |
206 | <execution> | 225 | <execution> |
207 | <id>thrift-sources</id> | 226 | <id>thrift-sources</id> |
208 | - <phase>generate-sources</phase> | 227 | + <phase>initialize</phase> |
209 | <goals> | 228 | <goals> |
210 | <goal>compile</goal> | 229 | <goal>compile</goal> |
211 | </goals> | 230 | </goals> |
... | @@ -227,13 +246,22 @@ | ... | @@ -227,13 +246,22 @@ |
227 | <configuration> | 246 | <configuration> |
228 | <sources> | 247 | <sources> |
229 | <source> | 248 | <source> |
230 | - ${project.build.directory}/generated-sources/thrift | 249 | + ${project.build.directory}/generated-sources |
231 | </source> | 250 | </source> |
232 | </sources> | 251 | </sources> |
233 | </configuration> | 252 | </configuration> |
234 | </execution> | 253 | </execution> |
235 | </executions> | 254 | </executions> |
236 | </plugin> | 255 | </plugin> |
256 | + <!-- OSGi --> | ||
257 | + <plugin> | ||
258 | + <groupId>org.apache.felix</groupId> | ||
259 | + <artifactId>maven-scr-plugin</artifactId> | ||
260 | + </plugin> | ||
261 | + <plugin> | ||
262 | + <groupId>org.onosproject</groupId> | ||
263 | + <artifactId>onos-maven-plugin</artifactId> | ||
264 | + </plugin> | ||
237 | </plugins> | 265 | </plugins> |
238 | </build> | 266 | </build> |
239 | 267 | ... | ... |
... | @@ -16,6 +16,8 @@ | ... | @@ -16,6 +16,8 @@ |
16 | 16 | ||
17 | package org.onosproject.bmv2.api.runtime; | 17 | package org.onosproject.bmv2.api.runtime; |
18 | 18 | ||
19 | +import org.onlab.util.ImmutableByteSequence; | ||
20 | + | ||
19 | import java.util.Collection; | 21 | import java.util.Collection; |
20 | 22 | ||
21 | /** | 23 | /** |
... | @@ -81,6 +83,15 @@ public interface Bmv2Client { | ... | @@ -81,6 +83,15 @@ public interface Bmv2Client { |
81 | String dumpTable(String tableName) throws Bmv2RuntimeException; | 83 | String dumpTable(String tableName) throws Bmv2RuntimeException; |
82 | 84 | ||
83 | /** | 85 | /** |
86 | + * Requests the device to transmit a given byte sequence over the given port. | ||
87 | + * | ||
88 | + * @param portNumber a port number | ||
89 | + * @param packet a byte sequence | ||
90 | + * @throws Bmv2RuntimeException | ||
91 | + */ | ||
92 | + void transmitPacket(int portNumber, ImmutableByteSequence packet) throws Bmv2RuntimeException; | ||
93 | + | ||
94 | + /** | ||
84 | * Reset the state of the switch (e.g. delete all entries, etc.). | 95 | * Reset the state of the switch (e.g. delete all entries, etc.). |
85 | * | 96 | * |
86 | * @throws Bmv2RuntimeException if any error occurs | 97 | * @throws Bmv2RuntimeException if any error occurs | ... | ... |
protocols/bmv2/src/main/java/org/onosproject/bmv2/api/runtime/Bmv2ControlPlaneServer.java
0 → 100644
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.bmv2.api.runtime; | ||
18 | + | ||
19 | +import org.onlab.util.ImmutableByteSequence; | ||
20 | + | ||
21 | +/** | ||
22 | + * A server that listens for requests from a BMv2 device. | ||
23 | + */ | ||
24 | +public interface Bmv2ControlPlaneServer { | ||
25 | + /** | ||
26 | + * Default listening port. | ||
27 | + */ | ||
28 | + int DEFAULT_PORT = 40123; | ||
29 | + | ||
30 | + /** | ||
31 | + * Register the given hello listener, to be called each time a hello message is received from a BMv2 device. | ||
32 | + * | ||
33 | + * @param listener a hello listener | ||
34 | + */ | ||
35 | + void addHelloListener(HelloListener listener); | ||
36 | + | ||
37 | + /** | ||
38 | + * Unregister the given hello listener. | ||
39 | + * | ||
40 | + * @param listener a hello listener | ||
41 | + */ | ||
42 | + void removeHelloListener(HelloListener listener); | ||
43 | + | ||
44 | + /** | ||
45 | + * Register the given packet listener, to be called each time a packet-in message is received from a BMv2 device. | ||
46 | + * | ||
47 | + * @param listener a packet listener | ||
48 | + */ | ||
49 | + void addPacketListener(PacketListener listener); | ||
50 | + | ||
51 | + /** | ||
52 | + * Unregister the given packet listener. | ||
53 | + * | ||
54 | + * @param listener a packet listener | ||
55 | + */ | ||
56 | + void removePacketListener(PacketListener listener); | ||
57 | + | ||
58 | + interface HelloListener { | ||
59 | + | ||
60 | + /** | ||
61 | + * Handles a hello message. | ||
62 | + * | ||
63 | + * @param device the BMv2 device that originated the message | ||
64 | + */ | ||
65 | + void handleHello(Bmv2Device device); | ||
66 | + } | ||
67 | + | ||
68 | + interface PacketListener { | ||
69 | + | ||
70 | + /** | ||
71 | + * Handles a packet-in message. | ||
72 | + * | ||
73 | + * @param device the BMv2 device that originated the message | ||
74 | + * @param inputPort the device port where the packet was received | ||
75 | + * @param reason a reason code | ||
76 | + * @param tableId the table id that originated this packet-in | ||
77 | + * @param contextId the context id where the packet-in was originated | ||
78 | + * @param packet the packet body | ||
79 | + */ | ||
80 | + void handlePacketIn(Bmv2Device device, int inputPort, long reason, int tableId, int contextId, | ||
81 | + ImmutableByteSequence packet); | ||
82 | + } | ||
83 | +} | ||
... | \ No newline at end of file | ... | \ No newline at end of file |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.bmv2.api.runtime; | ||
18 | + | ||
19 | +import com.google.common.base.Objects; | ||
20 | + | ||
21 | +import static com.google.common.base.Preconditions.checkNotNull; | ||
22 | + | ||
23 | +/** | ||
24 | + * Representation of a BMv2 device. | ||
25 | + */ | ||
26 | +public final class Bmv2Device { | ||
27 | + | ||
28 | + private final String thriftServerHost; | ||
29 | + private final int thriftServerPort; | ||
30 | + private final int internalDeviceId; | ||
31 | + | ||
32 | + /** | ||
33 | + * Creates a new Bmv2 device object. | ||
34 | + * | ||
35 | + * @param thriftServerHost the host of the Thrift runtime server running inside the device | ||
36 | + * @param thriftServerPort the port of the Thrift runtime server running inside the device | ||
37 | + * @param internalDeviceId the internal device id | ||
38 | + */ | ||
39 | + public Bmv2Device(String thriftServerHost, int thriftServerPort, int internalDeviceId) { | ||
40 | + this.thriftServerHost = checkNotNull(thriftServerHost, "host cannot be null"); | ||
41 | + this.thriftServerPort = checkNotNull(thriftServerPort, "port cannot be null"); | ||
42 | + this.internalDeviceId = internalDeviceId; | ||
43 | + } | ||
44 | + | ||
45 | + /** | ||
46 | + * Returns the hostname (or IP address) of the Thrift runtime server running inside the device. | ||
47 | + * | ||
48 | + * @return a string value | ||
49 | + */ | ||
50 | + public String thriftServerHost() { | ||
51 | + return thriftServerHost; | ||
52 | + } | ||
53 | + | ||
54 | + /** | ||
55 | + * Returns the port of the Thrift runtime server running inside the device. | ||
56 | + * | ||
57 | + * @return an integer value | ||
58 | + */ | ||
59 | + public int thriftServerPort() { | ||
60 | + return thriftServerPort; | ||
61 | + } | ||
62 | + | ||
63 | + /** | ||
64 | + * Returns the BMv2-internal device ID, which is an integer arbitrary chosen at device boot. | ||
65 | + * Such an ID must not be confused with the ONOS-internal {@link org.onosproject.net.DeviceId}. | ||
66 | + * | ||
67 | + * @return an integer value | ||
68 | + */ | ||
69 | + public int getInternalDeviceId() { | ||
70 | + return internalDeviceId; | ||
71 | + } | ||
72 | + | ||
73 | + @Override | ||
74 | + public int hashCode() { | ||
75 | + return Objects.hashCode(thriftServerHost, thriftServerPort, internalDeviceId); | ||
76 | + } | ||
77 | + | ||
78 | + @Override | ||
79 | + public boolean equals(Object obj) { | ||
80 | + if (this == obj) { | ||
81 | + return true; | ||
82 | + } | ||
83 | + if (obj == null || getClass() != obj.getClass()) { | ||
84 | + return false; | ||
85 | + } | ||
86 | + final Bmv2Device other = (Bmv2Device) obj; | ||
87 | + return Objects.equal(this.thriftServerHost, other.thriftServerHost) | ||
88 | + && Objects.equal(this.thriftServerPort, other.thriftServerPort) | ||
89 | + && Objects.equal(this.internalDeviceId, other.internalDeviceId); | ||
90 | + } | ||
91 | + | ||
92 | + @Override | ||
93 | + public String toString() { | ||
94 | + return thriftServerHost + ":" + thriftServerPort + "/" + internalDeviceId; | ||
95 | + } | ||
96 | +} |
1 | +/* | ||
2 | + * Copyright 2016-present Open Networking Laboratory | ||
3 | + * | ||
4 | + * Licensed under the Apache License, Version 2.0 (the "License"); | ||
5 | + * you may not use this file except in compliance with the License. | ||
6 | + * You may obtain a copy of the License at | ||
7 | + * | ||
8 | + * http://www.apache.org/licenses/LICENSE-2.0 | ||
9 | + * | ||
10 | + * Unless required by applicable law or agreed to in writing, software | ||
11 | + * distributed under the License is distributed on an "AS IS" BASIS, | ||
12 | + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
13 | + * See the License for the specific language governing permissions and | ||
14 | + * limitations under the License. | ||
15 | + */ | ||
16 | + | ||
17 | +package org.onosproject.bmv2.ctl; | ||
18 | + | ||
19 | +import com.google.common.collect.Maps; | ||
20 | +import org.apache.felix.scr.annotations.Activate; | ||
21 | +import org.apache.felix.scr.annotations.Component; | ||
22 | +import org.apache.felix.scr.annotations.Deactivate; | ||
23 | +import org.apache.felix.scr.annotations.Reference; | ||
24 | +import org.apache.felix.scr.annotations.ReferenceCardinality; | ||
25 | +import org.apache.felix.scr.annotations.Service; | ||
26 | +import org.apache.thrift.TException; | ||
27 | +import org.apache.thrift.TProcessor; | ||
28 | +import org.apache.thrift.protocol.TProtocol; | ||
29 | +import org.apache.thrift.server.TThreadPoolServer; | ||
30 | +import org.apache.thrift.transport.TServerSocket; | ||
31 | +import org.apache.thrift.transport.TServerTransport; | ||
32 | +import org.apache.thrift.transport.TSocket; | ||
33 | +import org.apache.thrift.transport.TTransportException; | ||
34 | +import org.onlab.util.ImmutableByteSequence; | ||
35 | +import org.onosproject.bmv2.api.runtime.Bmv2ControlPlaneServer; | ||
36 | +import org.onosproject.bmv2.api.runtime.Bmv2Device; | ||
37 | +import org.onosproject.core.CoreService; | ||
38 | +import org.p4.bmv2.thrift.ControlPlaneService; | ||
39 | +import org.slf4j.Logger; | ||
40 | +import org.slf4j.LoggerFactory; | ||
41 | + | ||
42 | +import java.nio.ByteBuffer; | ||
43 | +import java.util.Set; | ||
44 | +import java.util.concurrent.ConcurrentMap; | ||
45 | +import java.util.concurrent.CopyOnWriteArraySet; | ||
46 | +import java.util.concurrent.ExecutorService; | ||
47 | +import java.util.concurrent.Executors; | ||
48 | +import java.util.concurrent.TimeUnit; | ||
49 | + | ||
50 | +import static org.onlab.util.Tools.groupedThreads; | ||
51 | +import static org.p4.bmv2.thrift.ControlPlaneService.Processor; | ||
52 | + | ||
53 | +@Component(immediate = true) | ||
54 | +@Service | ||
55 | +public class Bmv2ControlPlaneThriftServer implements Bmv2ControlPlaneServer { | ||
56 | + | ||
57 | + private static final String APP_ID = "org.onosproject.bmv2"; | ||
58 | + private static final Logger LOG = LoggerFactory.getLogger(Bmv2ControlPlaneThriftServer.class); | ||
59 | + | ||
60 | + @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY) | ||
61 | + protected CoreService coreService; | ||
62 | + | ||
63 | + private final InternalTrackingProcessor trackingProcessor = new InternalTrackingProcessor(); | ||
64 | + private final ExecutorService executorService = Executors | ||
65 | + .newFixedThreadPool(16, groupedThreads("onos/bmv2", "control-plane-server", LOG)); | ||
66 | + | ||
67 | + private final Set<HelloListener> helloListeners = new CopyOnWriteArraySet<>(); | ||
68 | + private final Set<PacketListener> packetListeners = new CopyOnWriteArraySet<>(); | ||
69 | + | ||
70 | + private TThreadPoolServer thriftServer; | ||
71 | + private int serverPort = DEFAULT_PORT; | ||
72 | + | ||
73 | + @Activate | ||
74 | + public void activate() { | ||
75 | + coreService.registerApplication(APP_ID); | ||
76 | + try { | ||
77 | + TServerTransport transport = new TServerSocket(serverPort); | ||
78 | + LOG.info("Starting server on port {}...", serverPort); | ||
79 | + this.thriftServer = new TThreadPoolServer(new TThreadPoolServer.Args(transport) | ||
80 | + .processor(trackingProcessor) | ||
81 | + .executorService(executorService)); | ||
82 | + executorService.execute(thriftServer::serve); | ||
83 | + } catch (TTransportException e) { | ||
84 | + LOG.error("Unable to start server", e); | ||
85 | + } | ||
86 | + LOG.info("Activated"); | ||
87 | + } | ||
88 | + | ||
89 | + @Deactivate | ||
90 | + public void deactivate() { | ||
91 | + // Stop the server if running... | ||
92 | + if (thriftServer != null && !thriftServer.isServing()) { | ||
93 | + thriftServer.stop(); | ||
94 | + } | ||
95 | + try { | ||
96 | + executorService.awaitTermination(1, TimeUnit.SECONDS); | ||
97 | + } catch (InterruptedException e) { | ||
98 | + LOG.error("Server threads did not terminate"); | ||
99 | + } | ||
100 | + executorService.shutdownNow(); | ||
101 | + LOG.info("Deactivated"); | ||
102 | + } | ||
103 | + | ||
104 | + @Override | ||
105 | + public void addHelloListener(HelloListener listener) { | ||
106 | + if (!helloListeners.contains(listener)) { | ||
107 | + helloListeners.add(listener); | ||
108 | + } | ||
109 | + } | ||
110 | + | ||
111 | + @Override | ||
112 | + public void removeHelloListener(HelloListener listener) { | ||
113 | + helloListeners.remove(listener); | ||
114 | + } | ||
115 | + | ||
116 | + @Override | ||
117 | + public void addPacketListener(PacketListener listener) { | ||
118 | + if (!packetListeners.contains(listener)) { | ||
119 | + packetListeners.add(listener); | ||
120 | + } | ||
121 | + } | ||
122 | + | ||
123 | + @Override | ||
124 | + public void removePacketListener(PacketListener listener) { | ||
125 | + packetListeners.remove(listener); | ||
126 | + } | ||
127 | + | ||
128 | + /** | ||
129 | + * Handles service calls using registered listeners. | ||
130 | + */ | ||
131 | + private final class InternalServiceHandler implements ControlPlaneService.Iface { | ||
132 | + | ||
133 | + private final TSocket socket; | ||
134 | + private Bmv2Device remoteDevice; | ||
135 | + | ||
136 | + private InternalServiceHandler(TSocket socket) { | ||
137 | + this.socket = socket; | ||
138 | + } | ||
139 | + | ||
140 | + @Override | ||
141 | + public boolean ping() { | ||
142 | + return true; | ||
143 | + } | ||
144 | + | ||
145 | + @Override | ||
146 | + public void hello(int thriftServerPort, int deviceId) { | ||
147 | + // Locally note the remote device for future uses. | ||
148 | + String host = socket.getSocket().getInetAddress().getHostAddress(); | ||
149 | + remoteDevice = new Bmv2Device(host, thriftServerPort, deviceId); | ||
150 | + | ||
151 | + if (helloListeners.size() == 0) { | ||
152 | + LOG.debug("Received hello, but there's no listener registered."); | ||
153 | + } else { | ||
154 | + helloListeners.forEach(listener -> listener.handleHello(remoteDevice)); | ||
155 | + } | ||
156 | + } | ||
157 | + | ||
158 | + @Override | ||
159 | + public void packetIn(int port, long reason, int tableId, int contextId, ByteBuffer packet) { | ||
160 | + if (remoteDevice == null) { | ||
161 | + LOG.debug("Received packet-in, but the remote device is still unknown. Need a hello first..."); | ||
162 | + return; | ||
163 | + } | ||
164 | + | ||
165 | + if (packetListeners.size() == 0) { | ||
166 | + LOG.debug("Received packet-in, but there's no listener registered."); | ||
167 | + } else { | ||
168 | + packetListeners.forEach(listener -> listener.handlePacketIn(remoteDevice, | ||
169 | + port, | ||
170 | + reason, | ||
171 | + tableId, | ||
172 | + contextId, | ||
173 | + ImmutableByteSequence.copyFrom(packet))); | ||
174 | + } | ||
175 | + } | ||
176 | + } | ||
177 | + | ||
178 | + /** | ||
179 | + * Thrift Processor decorator. This class is needed in order to have access to the socket when handling a call. | ||
180 | + * Socket is needed to get the IP address of the client originating the call (see InternalServiceHandler.hello()) | ||
181 | + */ | ||
182 | + private final class InternalTrackingProcessor implements TProcessor { | ||
183 | + | ||
184 | + // Map sockets to processors. | ||
185 | + // TODO: implement it as a cache so unused sockets are expired automatically | ||
186 | + private final ConcurrentMap<TSocket, Processor<InternalServiceHandler>> processors = Maps.newConcurrentMap(); | ||
187 | + | ||
188 | + @Override | ||
189 | + public boolean process(final TProtocol in, final TProtocol out) throws TException { | ||
190 | + // Get the socket for this request. | ||
191 | + TSocket socket = (TSocket) in.getTransport(); | ||
192 | + // Get or create a processor for this socket | ||
193 | + Processor<InternalServiceHandler> processor = processors.computeIfAbsent(socket, s -> { | ||
194 | + InternalServiceHandler handler = new InternalServiceHandler(s); | ||
195 | + return new Processor<>(handler); | ||
196 | + }); | ||
197 | + // Delegate to the processor we are decorating. | ||
198 | + return processor.process(in, out); | ||
199 | + } | ||
200 | + } | ||
201 | +} |
... | @@ -31,6 +31,7 @@ import org.apache.thrift.protocol.TProtocol; | ... | @@ -31,6 +31,7 @@ import org.apache.thrift.protocol.TProtocol; |
31 | import org.apache.thrift.transport.TSocket; | 31 | import org.apache.thrift.transport.TSocket; |
32 | import org.apache.thrift.transport.TTransport; | 32 | import org.apache.thrift.transport.TTransport; |
33 | import org.apache.thrift.transport.TTransportException; | 33 | import org.apache.thrift.transport.TTransportException; |
34 | +import org.onlab.util.ImmutableByteSequence; | ||
34 | import org.onosproject.bmv2.api.runtime.Bmv2Action; | 35 | import org.onosproject.bmv2.api.runtime.Bmv2Action; |
35 | import org.onosproject.bmv2.api.runtime.Bmv2Client; | 36 | import org.onosproject.bmv2.api.runtime.Bmv2Client; |
36 | import org.onosproject.bmv2.api.runtime.Bmv2ExactMatchParam; | 37 | import org.onosproject.bmv2.api.runtime.Bmv2ExactMatchParam; |
... | @@ -50,6 +51,7 @@ import org.p4.bmv2.thrift.BmMatchParamTernary; | ... | @@ -50,6 +51,7 @@ import org.p4.bmv2.thrift.BmMatchParamTernary; |
50 | import org.p4.bmv2.thrift.BmMatchParamType; | 51 | import org.p4.bmv2.thrift.BmMatchParamType; |
51 | import org.p4.bmv2.thrift.BmMatchParamValid; | 52 | import org.p4.bmv2.thrift.BmMatchParamValid; |
52 | import org.p4.bmv2.thrift.DevMgrPortInfo; | 53 | import org.p4.bmv2.thrift.DevMgrPortInfo; |
54 | +import org.p4.bmv2.thrift.SimpleSwitch; | ||
53 | import org.p4.bmv2.thrift.Standard; | 55 | import org.p4.bmv2.thrift.Standard; |
54 | import org.slf4j.Logger; | 56 | import org.slf4j.Logger; |
55 | import org.slf4j.LoggerFactory; | 57 | import org.slf4j.LoggerFactory; |
... | @@ -88,15 +90,18 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -88,15 +90,18 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
88 | .expireAfterAccess(CLIENT_CACHE_TIMEOUT, TimeUnit.SECONDS) | 90 | .expireAfterAccess(CLIENT_CACHE_TIMEOUT, TimeUnit.SECONDS) |
89 | .removalListener(new ClientRemovalListener()) | 91 | .removalListener(new ClientRemovalListener()) |
90 | .build(new ClientLoader()); | 92 | .build(new ClientLoader()); |
91 | - private final Standard.Iface stdClient; | 93 | + private final Standard.Iface standardClient; |
94 | + private final SimpleSwitch.Iface simpleSwitchClient; | ||
92 | private final TTransport transport; | 95 | private final TTransport transport; |
93 | private final DeviceId deviceId; | 96 | private final DeviceId deviceId; |
94 | 97 | ||
95 | // ban constructor | 98 | // ban constructor |
96 | - private Bmv2ThriftClient(DeviceId deviceId, TTransport transport, Standard.Iface stdClient) { | 99 | + private Bmv2ThriftClient(DeviceId deviceId, TTransport transport, Standard.Iface standardClient, |
100 | + SimpleSwitch.Iface simpleSwitchClient) { | ||
97 | this.deviceId = deviceId; | 101 | this.deviceId = deviceId; |
98 | this.transport = transport; | 102 | this.transport = transport; |
99 | - this.stdClient = stdClient; | 103 | + this.standardClient = standardClient; |
104 | + this.simpleSwitchClient = simpleSwitchClient; | ||
100 | 105 | ||
101 | LOG.debug("New client created! > deviceId={}", deviceId); | 106 | LOG.debug("New client created! > deviceId={}", deviceId); |
102 | } | 107 | } |
... | @@ -131,9 +136,9 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -131,9 +136,9 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
131 | try { | 136 | try { |
132 | LOG.debug("Pinging device... > deviceId={}", deviceId); | 137 | LOG.debug("Pinging device... > deviceId={}", deviceId); |
133 | Bmv2ThriftClient client = of(deviceId); | 138 | Bmv2ThriftClient client = of(deviceId); |
134 | - client.stdClient.bm_dev_mgr_show_ports(); | 139 | + boolean result = client.simpleSwitchClient.ping(); |
135 | - LOG.debug("Device reachable! > deviceId={}", deviceId); | 140 | + LOG.debug("Device pinged! > deviceId={}, state={}", deviceId, result); |
136 | - return true; | 141 | + return result; |
137 | } catch (TException | Bmv2RuntimeException e) { | 142 | } catch (TException | Bmv2RuntimeException e) { |
138 | LOG.debug("Device NOT reachable! > deviceId={}", deviceId); | 143 | LOG.debug("Device NOT reachable! > deviceId={}", deviceId); |
139 | return false; | 144 | return false; |
... | @@ -242,7 +247,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -242,7 +247,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
242 | options.setPriority(entry.priority()); | 247 | options.setPriority(entry.priority()); |
243 | } | 248 | } |
244 | 249 | ||
245 | - entryId = stdClient.bm_mt_add_entry( | 250 | + entryId = standardClient.bm_mt_add_entry( |
246 | CONTEXT_ID, | 251 | CONTEXT_ID, |
247 | entry.tableName(), | 252 | entry.tableName(), |
248 | buildMatchParamsList(entry.matchKey()), | 253 | buildMatchParamsList(entry.matchKey()), |
... | @@ -253,7 +258,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -253,7 +258,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
253 | if (entry.hasTimeout()) { | 258 | if (entry.hasTimeout()) { |
254 | /* bmv2 accepts timeouts in milliseconds */ | 259 | /* bmv2 accepts timeouts in milliseconds */ |
255 | int msTimeout = (int) Math.round(entry.timeout() * 1_000); | 260 | int msTimeout = (int) Math.round(entry.timeout() * 1_000); |
256 | - stdClient.bm_mt_set_entry_ttl( | 261 | + standardClient.bm_mt_set_entry_ttl( |
257 | CONTEXT_ID, entry.tableName(), entryId, msTimeout); | 262 | CONTEXT_ID, entry.tableName(), entryId, msTimeout); |
258 | } | 263 | } |
259 | 264 | ||
... | @@ -285,7 +290,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -285,7 +290,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
285 | LOG.debug("Modifying table entry... > deviceId={}, entryId={}/{}", deviceId, tableName, entryId); | 290 | LOG.debug("Modifying table entry... > deviceId={}, entryId={}/{}", deviceId, tableName, entryId); |
286 | 291 | ||
287 | try { | 292 | try { |
288 | - stdClient.bm_mt_modify_entry( | 293 | + standardClient.bm_mt_modify_entry( |
289 | CONTEXT_ID, | 294 | CONTEXT_ID, |
290 | tableName, | 295 | tableName, |
291 | entryId, | 296 | entryId, |
... | @@ -306,7 +311,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -306,7 +311,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
306 | LOG.debug("Deleting table entry... > deviceId={}, entryId={}/{}", deviceId, tableName, entryId); | 311 | LOG.debug("Deleting table entry... > deviceId={}, entryId={}/{}", deviceId, tableName, entryId); |
307 | 312 | ||
308 | try { | 313 | try { |
309 | - stdClient.bm_mt_delete_entry(CONTEXT_ID, tableName, entryId); | 314 | + standardClient.bm_mt_delete_entry(CONTEXT_ID, tableName, entryId); |
310 | LOG.debug("Table entry deleted! > deviceId={}, entryId={}/{}", deviceId, tableName, entryId); | 315 | LOG.debug("Table entry deleted! > deviceId={}, entryId={}/{}", deviceId, tableName, entryId); |
311 | } catch (TException e) { | 316 | } catch (TException e) { |
312 | LOG.debug("Exception while deleting table entry: {} > deviceId={}, entryId={}/{}", | 317 | LOG.debug("Exception while deleting table entry: {} > deviceId={}, entryId={}/{}", |
... | @@ -322,7 +327,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -322,7 +327,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
322 | LOG.debug("Setting table default... > deviceId={}, tableName={}, action={}", deviceId, tableName, action); | 327 | LOG.debug("Setting table default... > deviceId={}, tableName={}, action={}", deviceId, tableName, action); |
323 | 328 | ||
324 | try { | 329 | try { |
325 | - stdClient.bm_mt_set_default_action( | 330 | + standardClient.bm_mt_set_default_action( |
326 | CONTEXT_ID, | 331 | CONTEXT_ID, |
327 | tableName, | 332 | tableName, |
328 | action.name(), | 333 | action.name(), |
... | @@ -341,7 +346,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -341,7 +346,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
341 | LOG.debug("Retrieving port info... > deviceId={}", deviceId); | 346 | LOG.debug("Retrieving port info... > deviceId={}", deviceId); |
342 | 347 | ||
343 | try { | 348 | try { |
344 | - List<DevMgrPortInfo> portInfos = stdClient.bm_dev_mgr_show_ports(); | 349 | + List<DevMgrPortInfo> portInfos = standardClient.bm_dev_mgr_show_ports(); |
345 | 350 | ||
346 | Collection<Bmv2PortInfo> bmv2PortInfos = Lists.newArrayList(); | 351 | Collection<Bmv2PortInfo> bmv2PortInfos = Lists.newArrayList(); |
347 | 352 | ||
... | @@ -366,7 +371,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -366,7 +371,7 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
366 | LOG.debug("Retrieving table dump... > deviceId={}, tableName={}", deviceId, tableName); | 371 | LOG.debug("Retrieving table dump... > deviceId={}, tableName={}", deviceId, tableName); |
367 | 372 | ||
368 | try { | 373 | try { |
369 | - String dump = stdClient.bm_dump_table(CONTEXT_ID, tableName); | 374 | + String dump = standardClient.bm_dump_table(CONTEXT_ID, tableName); |
370 | LOG.debug("Table dump retrieved! > deviceId={}, tableName={}", deviceId, tableName); | 375 | LOG.debug("Table dump retrieved! > deviceId={}, tableName={}", deviceId, tableName); |
371 | return dump; | 376 | return dump; |
372 | } catch (TException e) { | 377 | } catch (TException e) { |
... | @@ -377,12 +382,28 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -377,12 +382,28 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
377 | } | 382 | } |
378 | 383 | ||
379 | @Override | 384 | @Override |
385 | + public void transmitPacket(int portNumber, ImmutableByteSequence packet) throws Bmv2RuntimeException { | ||
386 | + | ||
387 | + LOG.debug("Requesting packet transmission... > portNumber={}, packet={}", portNumber, packet); | ||
388 | + | ||
389 | + try { | ||
390 | + | ||
391 | + simpleSwitchClient.push_packet(portNumber, ByteBuffer.wrap(packet.asArray())); | ||
392 | + LOG.debug("Packet transmission requested! > portNumber={}, packet={}", portNumber, packet); | ||
393 | + } catch (TException e) { | ||
394 | + LOG.debug("Exception while requesting packet transmission: {} > portNumber={}, packet={}", | ||
395 | + portNumber, packet); | ||
396 | + throw new Bmv2RuntimeException(e.getMessage(), e); | ||
397 | + } | ||
398 | + } | ||
399 | + | ||
400 | + @Override | ||
380 | public void resetState() throws Bmv2RuntimeException { | 401 | public void resetState() throws Bmv2RuntimeException { |
381 | 402 | ||
382 | LOG.debug("Resetting device state... > deviceId={}", deviceId); | 403 | LOG.debug("Resetting device state... > deviceId={}", deviceId); |
383 | 404 | ||
384 | try { | 405 | try { |
385 | - stdClient.bm_reset_state(); | 406 | + standardClient.bm_reset_state(); |
386 | LOG.debug("Device state reset! > deviceId={}", deviceId); | 407 | LOG.debug("Device state reset! > deviceId={}", deviceId); |
387 | } catch (TException e) { | 408 | } catch (TException e) { |
388 | LOG.debug("Exception while resetting device state: {} > deviceId={}", e, deviceId); | 409 | LOG.debug("Exception while resetting device state: {} > deviceId={}", e, deviceId); |
... | @@ -396,7 +417,6 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -396,7 +417,6 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
396 | private static class ClientLoader | 417 | private static class ClientLoader |
397 | extends CacheLoader<DeviceId, Bmv2ThriftClient> { | 418 | extends CacheLoader<DeviceId, Bmv2ThriftClient> { |
398 | 419 | ||
399 | - // Connection retries options: max 10 retries each 200 ms | ||
400 | private static final Options RECONN_OPTIONS = new Options(NUM_CONNECTION_RETRIES, TIME_BETWEEN_RETRIES); | 420 | private static final Options RECONN_OPTIONS = new Options(NUM_CONNECTION_RETRIES, TIME_BETWEEN_RETRIES); |
401 | 421 | ||
402 | @Override | 422 | @Override |
... | @@ -408,14 +428,20 @@ public final class Bmv2ThriftClient implements Bmv2Client { | ... | @@ -408,14 +428,20 @@ public final class Bmv2ThriftClient implements Bmv2Client { |
408 | TTransport transport = new TSocket( | 428 | TTransport transport = new TSocket( |
409 | info.getLeft(), info.getRight()); | 429 | info.getLeft(), info.getRight()); |
410 | TProtocol protocol = new TBinaryProtocol(transport); | 430 | TProtocol protocol = new TBinaryProtocol(transport); |
411 | - Standard.Client stdClient = new Standard.Client( | 431 | + // Our BMv2 device implements multiple Thrift services, create a client for each one. |
432 | + Standard.Client standardClient = new Standard.Client( | ||
412 | new TMultiplexedProtocol(protocol, "standard")); | 433 | new TMultiplexedProtocol(protocol, "standard")); |
413 | - // Wrap the client so to automatically have synchronization and resiliency to connectivity problems | 434 | + SimpleSwitch.Client simpleSwitch = new SimpleSwitch.Client( |
414 | - Standard.Iface reconnStdIface = SafeThriftClient.wrap(stdClient, | 435 | + new TMultiplexedProtocol(protocol, "simple_switch")); |
415 | - Standard.Iface.class, | 436 | + // Wrap clients so to automatically have synchronization and resiliency to connectivity errors |
416 | - RECONN_OPTIONS); | 437 | + Standard.Iface safeStandardClient = SafeThriftClient.wrap(standardClient, |
417 | - | 438 | + Standard.Iface.class, |
418 | - return new Bmv2ThriftClient(deviceId, transport, reconnStdIface); | 439 | + RECONN_OPTIONS); |
440 | + SimpleSwitch.Iface safeSimpleSwitchClient = SafeThriftClient.wrap(simpleSwitch, | ||
441 | + SimpleSwitch.Iface.class, | ||
442 | + RECONN_OPTIONS); | ||
443 | + | ||
444 | + return new Bmv2ThriftClient(deviceId, transport, safeStandardClient, safeSimpleSwitchClient); | ||
419 | } | 445 | } |
420 | } | 446 | } |
421 | 447 | ... | ... |
-
Please register or login to post a comment