2 * Copyright (c) 2015 Pantheon Technologies s.r.o. 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.openflowjava.protocol.impl.core.connection;
10 import com.google.common.base.Preconditions;
11 import java.net.InetSocketAddress;
12 import java.util.concurrent.TimeUnit;
13 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandler;
14 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
15 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.OfHeader;
16 import org.slf4j.Logger;
17 import org.slf4j.LoggerFactory;
19 final class OutboundQueueManager<T extends OutboundQueueHandler> extends AbstractOutboundQueueManager<T> {
20 private static final Logger LOG = LoggerFactory.getLogger(OutboundQueueManager.class);
22 private final int maxNonBarrierMessages;
23 private final long maxBarrierNanos;
25 // Updated from netty only
26 private boolean barrierTimerEnabled;
27 private long lastBarrierNanos = System.nanoTime();
28 private int nonBarrierMessages;
30 // Passed to executor to request a periodic barrier check
31 private final Runnable barrierRunnable = new Runnable() {
38 OutboundQueueManager(final ConnectionAdapterImpl parent, final InetSocketAddress address, final T handler,
39 final int maxNonBarrierMessages, final long maxBarrierNanos) {
40 super(parent, address, handler);
41 Preconditions.checkArgument(maxNonBarrierMessages > 0);
42 this.maxNonBarrierMessages = maxNonBarrierMessages;
43 Preconditions.checkArgument(maxBarrierNanos > 0);
44 this.maxBarrierNanos = maxBarrierNanos;
48 private void scheduleBarrierTimer(final long now) {
49 long next = lastBarrierNanos + maxBarrierNanos;
51 LOG.trace("Attempted to schedule barrier in the past, reset maximum)");
52 next = now + maxBarrierNanos;
55 final long delay = next - now;
56 LOG.trace("Scheduling barrier timer {}us from now", TimeUnit.NANOSECONDS.toMicros(delay));
57 parent.getChannel().eventLoop().schedule(barrierRunnable, next - now, TimeUnit.NANOSECONDS);
58 barrierTimerEnabled = true;
61 private void scheduleBarrierMessage() {
62 final Long xid = currentQueue.reserveBarrierIfNeeded();
64 LOG.trace("Queue {} already contains a barrier, not scheduling one", currentQueue);
68 currentQueue.commitEntry(xid, getHandler().createBarrierRequest(xid), null);
69 LOG.trace("Barrier XID {} scheduled", xid);
74 * Periodic barrier check.
76 protected void barrier() {
77 LOG.debug("Channel {} barrier timer expired", parent.getChannel());
78 barrierTimerEnabled = false;
80 LOG.trace("Channel shut down, not processing barrier");
84 final long now = System.nanoTime();
85 final long sinceLast = now - lastBarrierNanos;
86 if (sinceLast >= maxBarrierNanos) {
87 LOG.debug("Last barrier at {} now {}, elapsed {}", lastBarrierNanos, now, sinceLast);
88 // FIXME: we should be tracking requests/responses instead of this
89 if (nonBarrierMessages == 0) {
90 LOG.trace("No messages written since last barrier, not issuing one");
92 scheduleBarrierMessage();
98 * Write a message into the underlying channel.
100 * @param now Time reference for 'now'. We take this as an argument, as
101 * we need a timestamp to mark barrier messages we see swinging
102 * by. That timestamp does not need to be completely accurate,
103 * hence we use the flush start time. Alternative would be to
104 * measure System.nanoTime() for each barrier -- needlessly
108 void writeMessage(final OfHeader message, final long now) {
109 super.writeMessage(message, now);
110 if (message instanceof BarrierInput) {
111 LOG.trace("Barrier message seen, resetting counters");
112 nonBarrierMessages = 0;
113 lastBarrierNanos = now;
115 nonBarrierMessages++;
116 if (nonBarrierMessages >= maxNonBarrierMessages) {
117 LOG.trace("Scheduled barrier request after {} non-barrier messages", nonBarrierMessages);
118 scheduleBarrierMessage();
119 } else if (!barrierTimerEnabled) {
120 scheduleBarrierTimer(now);