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
20 AbstractOutboundQueueManager<T, StackedOutboundQueue> {
21 private static final Logger LOG = LoggerFactory.getLogger(OutboundQueueManager.class);
23 private final int maxNonBarrierMessages;
24 private final long maxBarrierNanos;
26 // Updated from netty only
27 private boolean barrierTimerEnabled;
28 private long lastBarrierNanos = System.nanoTime();
29 private int nonBarrierMessages;
31 // Passed to executor to request a periodic barrier check
32 private final Runnable barrierRunnable = this::barrier;
34 OutboundQueueManager(final ConnectionAdapterImpl parent, final InetSocketAddress address, final T handler,
35 final int maxNonBarrierMessages, final long maxBarrierNanos) {
36 super(parent, address, handler);
37 Preconditions.checkArgument(maxNonBarrierMessages > 0);
38 this.maxNonBarrierMessages = maxNonBarrierMessages;
39 Preconditions.checkArgument(maxBarrierNanos > 0);
40 this.maxBarrierNanos = maxBarrierNanos;
44 protected StackedOutboundQueue initializeStackedOutboudnqueue() {
45 return new StackedOutboundQueue(this);
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 if (currentQueue.isBarrierNeeded()) {
85 LOG.trace("Sending a barrier message");
86 scheduleBarrierMessage();
88 LOG.trace("Barrier not needed, not issuing one");
93 * Write a message into the underlying channel.
95 * @param now Time reference for 'now'. We take this as an argument, as
96 * we need a timestamp to mark barrier messages we see swinging
97 * by. That timestamp does not need to be completely accurate,
98 * hence we use the flush start time. Alternative would be to
99 * measure System.nanoTime() for each barrier -- needlessly
103 void writeMessage(final OfHeader message, final long now) {
104 super.writeMessage(message, now);
105 if (message instanceof BarrierInput) {
106 LOG.trace("Barrier message seen, resetting counters");
107 nonBarrierMessages = 0;
108 lastBarrierNanos = now;
110 nonBarrierMessages++;
111 if (nonBarrierMessages >= maxNonBarrierMessages) {
112 LOG.trace("Scheduled barrier request after {} non-barrier messages", nonBarrierMessages);
113 scheduleBarrierMessage();
114 } else if (!barrierTimerEnabled) {
115 scheduleBarrierTimer(now);