tom

Added IO loop test to the foo app.

...@@ -16,4 +16,20 @@ ...@@ -16,4 +16,20 @@
16 16
17 <description>ONOS application for miscellaneous experiments</description> 17 <description>ONOS application for miscellaneous experiments</description>
18 18
19 + <dependencies>
20 + <dependency>
21 + <groupId>org.onlab.onos</groupId>
22 + <artifactId>onos-cli</artifactId>
23 + <version>${project.version}</version>
24 + </dependency>
25 + <dependency>
26 + <groupId>org.onlab.onos</groupId>
27 + <artifactId>onlab-nio</artifactId>
28 + <version>${project.version}</version>
29 + </dependency>
30 + <dependency>
31 + <groupId>org.apache.karaf.shell</groupId>
32 + <artifactId>org.apache.karaf.shell.console</artifactId>
33 + </dependency>
34 + </dependencies>
19 </project> 35 </project>
......
1 +package org.onlab.onos.foo;
2 +
3 +import org.onlab.nio.IOLoop;
4 +import org.onlab.nio.MessageStream;
5 +import org.onlab.util.Counter;
6 +import org.slf4j.Logger;
7 +import org.slf4j.LoggerFactory;
8 +
9 +import java.io.IOException;
10 +import java.net.InetAddress;
11 +import java.net.InetSocketAddress;
12 +import java.net.SocketAddress;
13 +import java.nio.channels.ByteChannel;
14 +import java.nio.channels.SelectionKey;
15 +import java.nio.channels.SocketChannel;
16 +import java.text.DecimalFormat;
17 +import java.util.ArrayList;
18 +import java.util.List;
19 +import java.util.concurrent.ExecutionException;
20 +import java.util.concurrent.ExecutorService;
21 +import java.util.concurrent.Executors;
22 +import java.util.concurrent.FutureTask;
23 +import java.util.concurrent.Semaphore;
24 +import java.util.concurrent.TimeUnit;
25 +import java.util.concurrent.TimeoutException;
26 +
27 +import static java.lang.String.format;
28 +import static java.lang.System.out;
29 +import static org.onlab.onos.foo.IOLoopTestServer.PORT;
30 +import static org.onlab.util.Tools.delay;
31 +import static org.onlab.util.Tools.namedThreads;
32 +
33 +/**
34 + * Auxiliary test fixture to measure speed of NIO-based channels.
35 + */
36 +public class IOLoopTestClient {
37 +
38 + private static Logger log = LoggerFactory.getLogger(IOLoopTestClient.class);
39 +
40 + private final InetAddress ip;
41 + private final int port;
42 + private final int msgCount;
43 + private final int msgLength;
44 +
45 + private final List<CustomIOLoop> iloops = new ArrayList<>();
46 + private final ExecutorService ipool;
47 + private final ExecutorService wpool;
48 +
49 + Counter messages;
50 + Counter bytes;
51 +
52 + /**
53 + * Main entry point to launch the client.
54 + *
55 + * @param args command-line arguments
56 + * @throws java.io.IOException if unable to connect to server
57 + * @throws InterruptedException if latch wait gets interrupted
58 + * @throws java.util.concurrent.ExecutionException if wait gets interrupted
59 + * @throws java.util.concurrent.TimeoutException if timeout occurred while waiting for completion
60 + */
61 + public static void main(String[] args)
62 + throws IOException, InterruptedException, ExecutionException, TimeoutException {
63 + startStandalone(args);
64 +
65 + System.exit(0);
66 + }
67 +
68 + /**
69 + * Starts a standalone IO loop test client.
70 + *
71 + * @param args command-line arguments
72 + */
73 + public static void startStandalone(String[] args)
74 + throws IOException, InterruptedException, ExecutionException, TimeoutException {
75 + InetAddress ip = InetAddress.getByName(args.length > 0 ? args[0] : "127.0.0.1");
76 + int wc = args.length > 1 ? Integer.parseInt(args[1]) : 6;
77 + int mc = args.length > 2 ? Integer.parseInt(args[2]) : 50 * 1000000;
78 + int ml = args.length > 3 ? Integer.parseInt(args[3]) : 128;
79 + int to = args.length > 4 ? Integer.parseInt(args[4]) : 30;
80 +
81 + log.info("Setting up client with {} workers sending {} {}-byte messages to {} server... ",
82 + wc, mc, ml, ip);
83 + IOLoopTestClient client = new IOLoopTestClient(ip, wc, mc, ml, PORT);
84 +
85 + client.start();
86 + delay(500);
87 +
88 + client.await(to);
89 + client.report();
90 + }
91 +
92 + /**
93 + * Creates a speed client.
94 + *
95 + * @param ip ip address of server
96 + * @param wc worker count
97 + * @param mc message count to send per client
98 + * @param ml message length in bytes
99 + * @param port socket port
100 + * @throws java.io.IOException if unable to create IO loops
101 + */
102 + public IOLoopTestClient(InetAddress ip, int wc, int mc, int ml, int port) throws IOException {
103 + this.ip = ip;
104 + this.port = port;
105 + this.msgCount = mc;
106 + this.msgLength = ml;
107 + this.wpool = Executors.newFixedThreadPool(wc, namedThreads("worker"));
108 + this.ipool = Executors.newFixedThreadPool(wc, namedThreads("io-loop"));
109 +
110 + for (int i = 0; i < wc; i++) {
111 + iloops.add(new CustomIOLoop());
112 + }
113 + }
114 +
115 + /**
116 + * Starts the client workers.
117 + *
118 + * @throws java.io.IOException if unable to open connection
119 + */
120 + public void start() throws IOException {
121 + messages = new Counter();
122 + bytes = new Counter();
123 +
124 + // First start up all the IO loops
125 + for (CustomIOLoop l : iloops) {
126 + ipool.execute(l);
127 + }
128 +
129 + // Wait for all of them to get going
130 + for (CustomIOLoop l : iloops) {
131 + l.awaitStart(1000);
132 + }
133 +
134 + // ... and Next open all connections; one-per-loop
135 + for (CustomIOLoop l : iloops) {
136 + openConnection(l);
137 + }
138 + }
139 +
140 +
141 + /**
142 + * Initiates open connection request and registers the pending socket
143 + * channel with the given IO loop.
144 + *
145 + * @param loop loop with which the channel should be registered
146 + * @throws java.io.IOException if the socket could not be open or connected
147 + */
148 + private void openConnection(CustomIOLoop loop) throws IOException {
149 + SocketAddress sa = new InetSocketAddress(ip, port);
150 + SocketChannel ch = SocketChannel.open();
151 + ch.configureBlocking(false);
152 + loop.connectStream(ch);
153 + ch.connect(sa);
154 + }
155 +
156 +
157 + /**
158 + * Waits for the client workers to complete.
159 + *
160 + * @param secs timeout in seconds
161 + * @throws java.util.concurrent.ExecutionException if execution failed
162 + * @throws InterruptedException if interrupt occurred while waiting
163 + * @throws java.util.concurrent.TimeoutException if timeout occurred
164 + */
165 + public void await(int secs) throws InterruptedException,
166 + ExecutionException, TimeoutException {
167 + for (CustomIOLoop l : iloops) {
168 + if (l.worker.task != null) {
169 + l.worker.task.get(secs, TimeUnit.SECONDS);
170 + }
171 + }
172 + messages.freeze();
173 + bytes.freeze();
174 + }
175 +
176 + /**
177 + * Reports on the accumulated throughput trackers.
178 + */
179 + public void report() {
180 + DecimalFormat f = new DecimalFormat("#,##0");
181 + out.println(format("Client: %s messages; %s bytes; %s mps; %s Mbs",
182 + f.format(messages.total()), f.format(bytes.total()),
183 + f.format(messages.throughput()),
184 + f.format(bytes.throughput() / (1024 * msgLength))));
185 + }
186 +
187 +
188 + // Loop for transfer of fixed-length messages
189 + private class CustomIOLoop extends IOLoop<TestMessage, TestMessageStream> {
190 +
191 + Worker worker = new Worker();
192 +
193 + public CustomIOLoop() throws IOException {
194 + super(500);
195 + }
196 +
197 +
198 + @Override
199 + protected TestMessageStream createStream(ByteChannel channel) {
200 + return new TestMessageStream(msgLength, channel, this);
201 + }
202 +
203 + @Override
204 + protected synchronized void removeStream(MessageStream<TestMessage> stream) {
205 + super.removeStream(stream);
206 +
207 + messages.add(stream.messagesIn().total());
208 + bytes.add(stream.bytesIn().total());
209 +
210 +// out.println(format("Disconnected client; inbound %s mps, %s Mbps; outbound %s mps, %s Mbps",
211 +// FORMAT.format(stream.messagesIn().throughput()),
212 +// FORMAT.format(stream.bytesIn().throughput() / (1024 * msgLength)),
213 +// FORMAT.format(stream.messagesOut().throughput()),
214 +// FORMAT.format(stream.bytesOut().throughput() / (1024 * msgLength))));
215 +
216 + stream.messagesOut().reset();
217 + stream.bytesOut().reset();
218 + }
219 +
220 + @Override
221 + protected void processMessages(List<TestMessage> messages,
222 + MessageStream<TestMessage> b) {
223 + worker.release(messages.size());
224 + }
225 +
226 + @Override
227 + protected void connect(SelectionKey key) {
228 + super.connect(key);
229 + TestMessageStream b = (TestMessageStream) key.attachment();
230 + Worker w = ((CustomIOLoop) b.loop()).worker;
231 + w.pump(b);
232 + }
233 +
234 + }
235 +
236 + /**
237 + * Auxiliary worker to connect and pump batched messages using blocking I/O.
238 + */
239 + private class Worker implements Runnable {
240 +
241 + private static final int BATCH_SIZE = 1000;
242 + private static final int PERMITS = 2 * BATCH_SIZE;
243 +
244 + private TestMessageStream b;
245 + private FutureTask<Worker> task;
246 +
247 + // Stuff to throttle pump
248 + private final Semaphore semaphore = new Semaphore(PERMITS);
249 + private int msgWritten;
250 +
251 + void pump(TestMessageStream b) {
252 + this.b = b;
253 + task = new FutureTask<>(this, this);
254 + wpool.execute(task);
255 + }
256 +
257 + @Override
258 + public void run() {
259 + try {
260 + log.info("Worker started...");
261 +
262 + List<TestMessage> batch = new ArrayList<>();
263 + for (int i = 0; i < BATCH_SIZE; i++) {
264 + batch.add(new TestMessage(msgLength));
265 + }
266 +
267 + while (msgWritten < msgCount) {
268 + msgWritten += writeBatch(b, batch);
269 + }
270 +
271 + // Now try to get all the permits back before sending poison pill
272 + semaphore.acquireUninterruptibly(PERMITS);
273 + b.close();
274 +
275 + log.info("Worker done...");
276 +
277 + } catch (IOException e) {
278 + log.error("Worker unable to perform I/O", e);
279 + }
280 + }
281 +
282 +
283 + private int writeBatch(TestMessageStream b, List<TestMessage> batch)
284 + throws IOException {
285 + int count = Math.min(BATCH_SIZE, msgCount - msgWritten);
286 + acquire(count);
287 + if (count == BATCH_SIZE) {
288 + b.write(batch);
289 + } else {
290 + for (int i = 0; i < count; i++) {
291 + b.write(batch.get(i));
292 + }
293 + }
294 + return count;
295 + }
296 +
297 +
298 + // Release permits based on the specified number of message credits
299 + private void release(int permits) {
300 + semaphore.release(permits);
301 + }
302 +
303 + // Acquire permit for a single batch
304 + private void acquire(int permits) {
305 + semaphore.acquireUninterruptibly(permits);
306 + }
307 +
308 + }
309 +
310 +}
1 +package org.onlab.onos.foo;
2 +
3 +import org.onlab.nio.AcceptorLoop;
4 +import org.onlab.nio.IOLoop;
5 +import org.onlab.nio.MessageStream;
6 +import org.onlab.util.Counter;
7 +import org.slf4j.Logger;
8 +import org.slf4j.LoggerFactory;
9 +
10 +import java.io.IOException;
11 +import java.net.InetAddress;
12 +import java.net.InetSocketAddress;
13 +import java.net.Socket;
14 +import java.net.SocketAddress;
15 +import java.nio.channels.ByteChannel;
16 +import java.nio.channels.ServerSocketChannel;
17 +import java.nio.channels.SocketChannel;
18 +import java.text.DecimalFormat;
19 +import java.util.ArrayList;
20 +import java.util.List;
21 +import java.util.concurrent.ExecutorService;
22 +import java.util.concurrent.Executors;
23 +
24 +import static java.lang.String.format;
25 +import static java.lang.System.out;
26 +import static org.onlab.util.Tools.delay;
27 +import static org.onlab.util.Tools.namedThreads;
28 +
29 +/**
30 + * Auxiliary test fixture to measure speed of NIO-based channels.
31 + */
32 +public class IOLoopTestServer {
33 +
34 + private static Logger log = LoggerFactory.getLogger(IOLoopTestServer.class);
35 +
36 + private static final int PRUNE_FREQUENCY = 1000;
37 +
38 + static final int PORT = 9876;
39 + static final long TIMEOUT = 1000;
40 +
41 + static final boolean SO_NO_DELAY = false;
42 + static final int SO_SEND_BUFFER_SIZE = 128 * 1024;
43 + static final int SO_RCV_BUFFER_SIZE = 128 * 1024;
44 +
45 + static final DecimalFormat FORMAT = new DecimalFormat("#,##0");
46 +
47 + private final AcceptorLoop aloop;
48 + private final ExecutorService apool = Executors.newSingleThreadExecutor(namedThreads("accept"));
49 +
50 + private final List<CustomIOLoop> iloops = new ArrayList<>();
51 + private final ExecutorService ipool;
52 +
53 + private final int workerCount;
54 + private final int msgLength;
55 + private int lastWorker = -1;
56 +
57 + Counter messages;
58 + Counter bytes;
59 +
60 + /**
61 + * Main entry point to launch the server.
62 + *
63 + * @param args command-line arguments
64 + * @throws java.io.IOException if unable to crate IO loops
65 + */
66 + public static void main(String[] args) throws IOException {
67 + startStandalone(args);
68 + System.exit(0);
69 + }
70 +
71 + /**
72 + * Starts a standalone IO loop test server.
73 + *
74 + * @param args command-line arguments
75 + */
76 + public static void startStandalone(String[] args) throws IOException {
77 + InetAddress ip = InetAddress.getByName(args.length > 0 ? args[0] : "127.0.0.1");
78 + int wc = args.length > 1 ? Integer.parseInt(args[1]) : 6;
79 + int ml = args.length > 2 ? Integer.parseInt(args[2]) : 128;
80 +
81 + log.info("Setting up the server with {} workers, {} byte messages on {}... ",
82 + wc, ml, ip);
83 + IOLoopTestServer server = new IOLoopTestServer(ip, wc, ml, PORT);
84 + server.start();
85 +
86 + // Start pruning clients.
87 + while (true) {
88 + delay(PRUNE_FREQUENCY);
89 + server.prune();
90 + }
91 + }
92 +
93 + /**
94 + * Creates a speed server.
95 + *
96 + * @param ip optional ip of the adapter where to bind
97 + * @param wc worker count
98 + * @param ml message length in bytes
99 + * @param port listen port
100 + * @throws java.io.IOException if unable to create IO loops
101 + */
102 + public IOLoopTestServer(InetAddress ip, int wc, int ml, int port) throws IOException {
103 + this.workerCount = wc;
104 + this.msgLength = ml;
105 + this.ipool = Executors.newFixedThreadPool(workerCount, namedThreads("io-loop"));
106 +
107 + this.aloop = new CustomAcceptLoop(new InetSocketAddress(ip, port));
108 + for (int i = 0; i < workerCount; i++) {
109 + iloops.add(new CustomIOLoop());
110 + }
111 + }
112 +
113 + /**
114 + * Start the server IO loops and kicks off throughput tracking.
115 + */
116 + public void start() {
117 + messages = new Counter();
118 + bytes = new Counter();
119 +
120 + for (CustomIOLoop l : iloops) {
121 + ipool.execute(l);
122 + }
123 + apool.execute(aloop);
124 +
125 + for (CustomIOLoop l : iloops) {
126 + l.awaitStart(TIMEOUT);
127 + }
128 + aloop.awaitStart(TIMEOUT);
129 + }
130 +
131 + /**
132 + * Stop the server IO loops and freezes throughput tracking.
133 + */
134 + public void stop() {
135 + aloop.shutdown();
136 + for (CustomIOLoop l : iloops) {
137 + l.shutdown();
138 + }
139 +
140 + for (CustomIOLoop l : iloops) {
141 + l.awaitStop(TIMEOUT);
142 + }
143 + aloop.awaitStop(TIMEOUT);
144 +
145 + messages.freeze();
146 + bytes.freeze();
147 + }
148 +
149 + /**
150 + * Reports on the accumulated throughput trackers.
151 + */
152 + public void report() {
153 + DecimalFormat f = new DecimalFormat("#,##0");
154 + out.println(format("Server: %s messages; %s bytes; %s mps; %s Mbs",
155 + f.format(messages.total()), f.format(bytes.total()),
156 + f.format(messages.throughput()),
157 + f.format(bytes.throughput() / (1024 * msgLength))));
158 + }
159 +
160 + /**
161 + * Prunes the IO loops of stale message buffers.
162 + */
163 + public void prune() {
164 + for (CustomIOLoop l : iloops) {
165 + l.pruneStaleStreams();
166 + }
167 + }
168 +
169 + // Get the next worker to which a client should be assigned
170 + private synchronized CustomIOLoop nextWorker() {
171 + lastWorker = (lastWorker + 1) % workerCount;
172 + return iloops.get(lastWorker);
173 + }
174 +
175 + // Loop for transfer of fixed-length messages
176 + private class CustomIOLoop extends IOLoop<TestMessage, TestMessageStream> {
177 +
178 + public CustomIOLoop() throws IOException {
179 + super(500);
180 + }
181 +
182 + @Override
183 + protected TestMessageStream createStream(ByteChannel channel) {
184 + return new TestMessageStream(msgLength, channel, this);
185 + }
186 +
187 + @Override
188 + protected void removeStream(MessageStream<TestMessage> stream) {
189 + super.removeStream(stream);
190 +
191 + messages.add(stream.messagesIn().total());
192 + bytes.add(stream.bytesIn().total());
193 +
194 +// out.println(format("Disconnected server; inbound %s mps, %s Mbps; outbound %s mps, %s Mbps",
195 +// FORMAT.format(stream.messagesIn().throughput()),
196 +// FORMAT.format(stream.bytesIn().throughput() / (1024 * msgLength)),
197 +// FORMAT.format(stream.messagesOut().throughput()),
198 +// FORMAT.format(stream.bytesOut().throughput() / (1024 * msgLength))));
199 + }
200 +
201 + @Override
202 + protected void processMessages(List<TestMessage> messages,
203 + MessageStream<TestMessage> stream) {
204 + try {
205 + stream.write(messages);
206 + } catch (IOException e) {
207 + log.error("Unable to echo messages", e);
208 + }
209 + }
210 + }
211 +
212 + // Loop for accepting client connections
213 + private class CustomAcceptLoop extends AcceptorLoop {
214 +
215 + public CustomAcceptLoop(SocketAddress address) throws IOException {
216 + super(500, address);
217 + }
218 +
219 + @Override
220 + protected void acceptConnection(ServerSocketChannel channel) throws IOException {
221 + SocketChannel sc = channel.accept();
222 + sc.configureBlocking(false);
223 +
224 + Socket so = sc.socket();
225 + so.setTcpNoDelay(SO_NO_DELAY);
226 + so.setReceiveBufferSize(SO_RCV_BUFFER_SIZE);
227 + so.setSendBufferSize(SO_SEND_BUFFER_SIZE);
228 +
229 + nextWorker().acceptStream(sc);
230 + log.info("Connected client");
231 + }
232 + }
233 +
234 +}
1 +package org.onlab.onos.foo;
2 +
3 +import org.apache.karaf.shell.commands.Command;
4 +import org.onlab.onos.cli.AbstractShellCommand;
5 +
6 +import static org.onlab.onos.foo.IOLoopTestClient.startStandalone;
7 +
8 +/**
9 + * Starts the test IO loop client.
10 + */
11 +@Command(scope = "onos", name = "test-io-client",
12 + description = "Starts the test IO loop client")
13 +public class TestIOClientCommand extends AbstractShellCommand {
14 +
15 + @Override
16 + protected void execute() {
17 + try {
18 + startStandalone(new String[]{});
19 + } catch (Exception e) {
20 + error("Unable to start server %s", e);
21 + }
22 + }
23 +
24 +}
1 +package org.onlab.onos.foo;
2 +
3 +import org.apache.karaf.shell.commands.Command;
4 +import org.onlab.onos.cli.AbstractShellCommand;
5 +
6 +import static org.onlab.onos.foo.IOLoopTestServer.startStandalone;
7 +
8 +
9 +/**
10 + * Starts the test IO loop server.
11 + */
12 +@Command(scope = "onos", name = "test-io-server",
13 + description = "Starts the test IO loop server")
14 +public class TestIOServerCommand extends AbstractShellCommand {
15 +
16 + @Override
17 + protected void execute() {
18 + try {
19 + startStandalone(new String[]{});
20 + } catch (Exception e) {
21 + error("Unable to start server %s", e);
22 + }
23 + }
24 +
25 +}
1 +package org.onlab.onos.foo;
2 +
3 +import org.onlab.nio.AbstractMessage;
4 +
5 +/**
6 + * Fixed-length message.
7 + */
8 +public class TestMessage extends AbstractMessage {
9 +
10 + private final byte[] data;
11 +
12 + /**
13 + * Creates a new message with the specified length.
14 + *
15 + * @param length message length
16 + */
17 + public TestMessage(int length) {
18 + this.length = length;
19 + data = new byte[length];
20 + }
21 +
22 + /**
23 + * Creates a new message with the specified data.
24 + *
25 + * @param data message data
26 + */
27 + TestMessage(byte[] data) {
28 + this.length = data.length;
29 + this.data = data;
30 + }
31 +
32 + /**
33 + * Gets the backing byte array data.
34 + *
35 + * @return backing byte array
36 + */
37 + public byte[] data() {
38 + return data;
39 + }
40 +
41 +}
1 +package org.onlab.onos.foo;
2 +
3 +import org.onlab.nio.IOLoop;
4 +import org.onlab.nio.MessageStream;
5 +
6 +import java.nio.ByteBuffer;
7 +import java.nio.channels.ByteChannel;
8 +
9 +/**
10 + * Fixed-length message transfer buffer.
11 + */
12 +public class TestMessageStream extends MessageStream<TestMessage> {
13 +
14 + private static final String E_WRONG_LEN = "Illegal message length: ";
15 +
16 + private final int length;
17 +
18 + /**
19 + * Create a new buffer for transferring messages of the specified length.
20 + *
21 + * @param length message length
22 + * @param ch backing channel
23 + * @param loop driver loop
24 + */
25 + public TestMessageStream(int length, ByteChannel ch,
26 + IOLoop<TestMessage, ?> loop) {
27 + super(loop, ch, 64 * 1024, 500);
28 + this.length = length;
29 + }
30 +
31 + @Override
32 + protected TestMessage read(ByteBuffer rb) {
33 + if (rb.remaining() < length) {
34 + return null;
35 + }
36 + TestMessage message = new TestMessage(length);
37 + rb.get(message.data());
38 + return message;
39 + }
40 +
41 + /**
42 + * {@inheritDoc}
43 + * <p/>
44 + * This implementation enforces the message length against the buffer
45 + * supported length.
46 + *
47 + * @throws IllegalArgumentException if message size does not match the
48 + * supported buffer size
49 + */
50 + @Override
51 + protected void write(TestMessage message, ByteBuffer wb) {
52 + if (message.length() != length) {
53 + throw new IllegalArgumentException(E_WRONG_LEN + message.length());
54 + }
55 + wb.put(message.data());
56 + }
57 +
58 +}
1 +<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0">
2 +
3 + <command-bundle xmlns="http://karaf.apache.org/xmlns/shell/v1.1.0">
4 + <command>
5 + <action class="org.onlab.onos.foo.TestIOClientCommand"/>
6 + </command>
7 + <command>
8 + <action class="org.onlab.onos.foo.TestIOServerCommand"/>
9 + </command>
10 + </command-bundle>
11 +
12 +</blueprint>
...@@ -17,6 +17,8 @@ ...@@ -17,6 +17,8 @@
17 <bundle>mvn:com.esotericsoftware/minlog/1.3.0</bundle> 17 <bundle>mvn:com.esotericsoftware/minlog/1.3.0</bundle>
18 <bundle>mvn:org.objenesis/objenesis/2.1</bundle> 18 <bundle>mvn:org.objenesis/objenesis/2.1</bundle>
19 <bundle>mvn:de.javakaffee/kryo-serializers/0.27</bundle> 19 <bundle>mvn:de.javakaffee/kryo-serializers/0.27</bundle>
20 +
21 + <bundle>mvn:org.onlab.onos/onlab-nio/1.0.0-SNAPSHOT</bundle>
20 </feature> 22 </feature>
21 23
22 <feature name="onos-thirdparty-web" version="1.0.0" 24 <feature name="onos-thirdparty-web" version="1.0.0"
......
...@@ -172,6 +172,11 @@ ...@@ -172,6 +172,11 @@
172 </dependency> 172 </dependency>
173 <dependency> 173 <dependency>
174 <groupId>org.onlab.onos</groupId> 174 <groupId>org.onlab.onos</groupId>
175 + <artifactId>onlab-nio</artifactId>
176 + <version>${project.version}</version>
177 + </dependency>
178 + <dependency>
179 + <groupId>org.onlab.onos</groupId>
175 <artifactId>onlab-osgi</artifactId> 180 <artifactId>onlab-osgi</artifactId>
176 <version>${project.version}</version> 181 <version>${project.version}</version>
177 </dependency> 182 </dependency>
......
...@@ -52,4 +52,18 @@ public abstract class Tools { ...@@ -52,4 +52,18 @@ public abstract class Tools {
52 public static String toHex(long value, int width) { 52 public static String toHex(long value, int width) {
53 return Strings.padStart(UnsignedLongs.toString(value, 16), width, '0'); 53 return Strings.padStart(UnsignedLongs.toString(value, 16), width, '0');
54 } 54 }
55 +
56 + /**
57 + * Suspends the current thread for a specified number of millis.
58 + *
59 + * @param ms number of millis
60 + */
61 + public static void delay(int ms) {
62 + try {
63 + Thread.sleep(ms);
64 + } catch (InterruptedException e) {
65 + throw new RuntimeException("Interrupted", e);
66 + }
67 + }
68 +
55 } 69 }
......
...@@ -66,7 +66,7 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>> ...@@ -66,7 +66,7 @@ public abstract class IOLoop<M extends Message, S extends MessageStream<M>>
66 * 66 *
67 * @param stream message stream to remove 67 * @param stream message stream to remove
68 */ 68 */
69 - void removeStream(MessageStream<M> stream) { 69 + protected void removeStream(MessageStream<M> stream) {
70 streams.remove(stream); 70 streams.remove(stream);
71 } 71 }
72 72
......
...@@ -24,8 +24,8 @@ import java.util.concurrent.TimeoutException; ...@@ -24,8 +24,8 @@ import java.util.concurrent.TimeoutException;
24 24
25 import static java.lang.String.format; 25 import static java.lang.String.format;
26 import static java.lang.System.out; 26 import static java.lang.System.out;
27 -import static org.onlab.junit.TestTools.delay;
28 import static org.onlab.nio.IOLoopTestServer.PORT; 27 import static org.onlab.nio.IOLoopTestServer.PORT;
28 +import static org.onlab.util.Tools.delay;
29 import static org.onlab.util.Tools.namedThreads; 29 import static org.onlab.util.Tools.namedThreads;
30 30
31 /** 31 /**
...@@ -81,7 +81,7 @@ public class IOLoopTestClient { ...@@ -81,7 +81,7 @@ public class IOLoopTestClient {
81 IOLoopTestClient client = new IOLoopTestClient(ip, wc, mc, ml, PORT); 81 IOLoopTestClient client = new IOLoopTestClient(ip, wc, mc, ml, PORT);
82 82
83 client.start(); 83 client.start();
84 - delay(2000); 84 + delay(500);
85 85
86 client.await(to); 86 client.await(to);
87 client.report(); 87 client.report();
......
...@@ -70,7 +70,7 @@ public class IOLoopTestServer { ...@@ -70,7 +70,7 @@ public class IOLoopTestServer {
70 * 70 *
71 * @param args command-line arguments 71 * @param args command-line arguments
72 */ 72 */
73 - private static void startStandalone(String[] args) throws IOException { 73 + public static void startStandalone(String[] args) throws IOException {
74 InetAddress ip = InetAddress.getByName(args.length > 0 ? args[0] : "127.0.0.1"); 74 InetAddress ip = InetAddress.getByName(args.length > 0 ? args[0] : "127.0.0.1");
75 int wc = args.length > 1 ? Integer.parseInt(args[1]) : 6; 75 int wc = args.length > 1 ? Integer.parseInt(args[1]) : 6;
76 int ml = args.length > 2 ? Integer.parseInt(args[2]) : 128; 76 int ml = args.length > 2 ? Integer.parseInt(args[2]) : 128;
......