2 * Copyright (c) 2014 Cisco Systems, Inc. and others. All rights reserved.
4 * This program and the accompanying materials are made available under the
5 * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6 * and is available at http://www.eclipse.org/legal/epl-v10.html
8 package org.opendaylight.openflowplugin.testcommon;
10 import static org.opendaylight.openflowjava.util.ByteBufUtils.macAddressToString;
12 import com.google.common.util.concurrent.ThreadFactoryBuilder;
13 import java.util.Arrays;
14 import java.util.Collections;
15 import java.util.List;
16 import java.util.concurrent.ArrayBlockingQueue;
17 import java.util.concurrent.ExecutorService;
18 import java.util.concurrent.RejectedExecutionException;
19 import java.util.concurrent.RejectedExecutionHandler;
20 import java.util.concurrent.ThreadPoolExecutor;
21 import java.util.concurrent.TimeUnit;
22 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
23 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.yang.types.rev100924.MacAddress;
24 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.DropActionCaseBuilder;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.drop.action._case.DropAction;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.action.drop.action._case.DropActionBuilder;
27 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.Action;
28 import org.opendaylight.yang.gen.v1.urn.opendaylight.action.types.rev131112.action.list.ActionBuilder;
29 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.Instructions;
30 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.InstructionsBuilder;
31 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.Match;
32 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.flow.MatchBuilder;
33 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.instruction.ApplyActionsCaseBuilder;
34 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.instruction.apply.actions._case.ApplyActionsBuilder;
35 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.Instruction;
36 import org.opendaylight.yang.gen.v1.urn.opendaylight.flow.types.rev131026.instruction.list.InstructionBuilder;
37 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.NodeConnectorRef;
38 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.Node;
39 import org.opendaylight.yang.gen.v1.urn.opendaylight.inventory.rev130819.nodes.NodeKey;
40 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.match.types.rev131026.ethernet.match.fields.EthernetSourceBuilder;
41 import org.opendaylight.yang.gen.v1.urn.opendaylight.model.match.types.rev131026.match.EthernetMatchBuilder;
42 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketProcessingListener;
43 import org.opendaylight.yang.gen.v1.urn.opendaylight.packet.service.rev130709.PacketReceived;
44 import org.opendaylight.yangtools.yang.binding.InstanceIdentifier;
45 import org.slf4j.Logger;
46 import org.slf4j.LoggerFactory;
48 abstract class AbstractDropTest implements PacketProcessingListener, AutoCloseable {
49 private static final Logger LOG = LoggerFactory.getLogger(AbstractDropTest.class);
51 protected static final Integer PRIORITY = 4;
52 protected static final Long BUFFER_ID = 0L;
53 protected static final Integer HARD_TIMEOUT = 300;
54 protected static final Integer IDLE_TIMEOUT = 240;
55 protected static final short TABLE_ID = 0;
57 static final long STARTUP_LOOP_TICK = 500L;
58 static final int STARTUP_LOOP_MAX_RETRIES = 8;
59 private static final int PROCESSING_POOL_SIZE = 10000;
61 private final int POOL_THREAD_AMOUNT = 8;
62 private final ExecutorService executorService;
65 private static final AtomicIntegerFieldUpdater<AbstractDropTest> SENT_UPDATER = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "sent");
66 private volatile int sent;
68 private static final AtomicIntegerFieldUpdater<AbstractDropTest> RCVD_UPDATER = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "rcvd");
69 private volatile int rcvd;
71 private static final AtomicIntegerFieldUpdater<AbstractDropTest> EXCS_UPDATER = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "excs");
72 private volatile int excs;
74 protected static final AtomicIntegerFieldUpdater<AbstractDropTest> RPC_FUTURE_SUCCESS_UPDATER = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "ftrSuccess");
75 protected volatile int ftrSuccess;
77 protected static final AtomicIntegerFieldUpdater<AbstractDropTest> RPC_FUTURE_FAIL_UPDATER = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "ftrFailed");
78 protected volatile int ftrFailed;
80 protected static final AtomicIntegerFieldUpdater<AbstractDropTest> RUNABLES_EXECUTED = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "runablesExecuted");
81 protected volatile int runablesExecuted;
83 protected static final AtomicIntegerFieldUpdater<AbstractDropTest> RUNABLES_REJECTED = AtomicIntegerFieldUpdater.newUpdater(AbstractDropTest.class, "runablesRejected");
84 protected volatile int runablesRejected;
86 public final DropTestStats getStats() {
87 return new DropTestStats(this.sent, this.rcvd, this.excs, this.ftrFailed, this.ftrSuccess, this.runablesExecuted, this.runablesRejected);
90 public AbstractDropTest() {
91 final ArrayBlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<Runnable>(PROCESSING_POOL_SIZE);
92 ThreadPoolExecutor threadPool = new ThreadPoolExecutor(POOL_THREAD_AMOUNT, POOL_THREAD_AMOUNT, 0,
93 TimeUnit.MILLISECONDS,
95 threadPool.setThreadFactory(new ThreadFactoryBuilder().setNameFormat("dropTest-%d").build());
96 threadPool.setRejectedExecutionHandler(new RejectedExecutionHandler() {
98 public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
101 } catch (InterruptedException e) {
102 throw new RejectedExecutionException("Interrupted while waiting on queue", e);
107 executorService = threadPool;
110 public final void clearStats() {
116 this.runablesExecuted = 0;
117 this.runablesRejected = 0;
120 private final void incrementRunableExecuted() {
121 RUNABLES_EXECUTED.incrementAndGet(this);
124 private final void incrementRunableRejected() {
125 RUNABLES_REJECTED.incrementAndGet(this);
129 public final void onPacketReceived(final PacketReceived notification) {
130 LOG.debug("onPacketReceived - Entering - {}", notification);
132 RCVD_UPDATER.incrementAndGet(this);
135 executorService.submit(new Runnable() {
138 incrementRunableExecuted();
139 processPacket(notification);
142 } catch (Exception e) {
143 incrementRunableRejected();
145 LOG.debug("onPacketReceived - Leaving", notification);
148 private void processPacket(final PacketReceived notification) {
150 // TODO Auto-generated method stub
151 final byte[] rawPacket = notification.getPayload();
152 final byte[] srcMac = Arrays.copyOfRange(rawPacket, 6, 12);
154 final MatchBuilder match = new MatchBuilder();
155 final EthernetMatchBuilder ethernetMatch = new EthernetMatchBuilder();
156 final EthernetSourceBuilder ethSourceBuilder = new EthernetSourceBuilder();
158 //TODO: use HEX, use binary form
159 //Hex.decodeHex("000000000001".toCharArray());
161 ethSourceBuilder.setAddress(new MacAddress(macAddressToString(srcMac)));
162 ethernetMatch.setEthernetSource(ethSourceBuilder.build());
163 match.setEthernetMatch(ethernetMatch.build());
165 final DropActionBuilder dab = new DropActionBuilder();
166 final DropAction dropAction = dab.build();
167 final ActionBuilder ab = new ActionBuilder();
169 ab.setAction(new DropActionCaseBuilder().setDropAction(dropAction).build());
171 // Add our drop action to a list
172 final List<Action> actionList = Collections.singletonList(ab.build());
174 // Create an Apply Action
175 final ApplyActionsBuilder aab = new ApplyActionsBuilder();
176 aab.setAction(actionList);
178 // Wrap our Apply Action in an Instruction
179 final InstructionBuilder ib = new InstructionBuilder();
180 ib.setInstruction(new ApplyActionsCaseBuilder().setApplyActions(aab.build()).build()).setOrder(0);
182 // Put our Instruction in a list of Instructions
183 final InstructionsBuilder isb = new InstructionsBuilder();
184 final List<Instruction> instructions = Collections.singletonList(ib.build());
185 isb.setInstruction(instructions);
187 // Get the Ingress nodeConnectorRef
188 final NodeConnectorRef ncr = notification.getIngress();
190 // Get the instance identifier for the nodeConnectorRef
191 final InstanceIdentifier<?> ncri = ncr.getValue();
193 // Get the instanceID for the Node in the tree above us
194 final NodeKey node = ncri.firstKeyOf(Node.class, NodeKey.class);
196 processPacket(node, match.build(), isb.build());
198 SENT_UPDATER.incrementAndGet(this);
199 } catch (final Exception e) {
200 // TODO Auto-generated catch block
201 LOG.warn("Failed to process packet: {}", e.getMessage());
202 LOG.debug("Failed to process packet.. ", e);
203 EXCS_UPDATER.incrementAndGet(this);
207 protected abstract void processPacket(NodeKey node, Match match, Instructions instructions);
211 public void close() {
212 executorService.shutdown();
215 public void countFutureSuccess() {
216 RPC_FUTURE_SUCCESS_UPDATER.incrementAndGet(this);
219 public void countFutureError() {
220 RPC_FUTURE_FAIL_UPDATER.incrementAndGet(this);