Bug 6585: BGP ChannelOutputLimiter waits forever
[bgpcep.git] / bgp / rib-impl / src / main / java / org / opendaylight / protocol / bgp / rib / impl / ChannelOutputLimiter.java
1 /*
2  * Copyright (c) 2015 Cisco Systems, Inc. and others.  All rights reserved.
3  *
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
7  */
8 package org.opendaylight.protocol.bgp.rib.impl;
9
10 import com.google.common.base.Preconditions;
11 import io.netty.channel.ChannelFuture;
12 import io.netty.channel.ChannelHandlerContext;
13 import io.netty.channel.ChannelInboundHandlerAdapter;
14 import javax.annotation.concurrent.ThreadSafe;
15 import org.opendaylight.yangtools.yang.binding.Notification;
16 import org.slf4j.Logger;
17 import org.slf4j.LoggerFactory;
18
19 /**
20  * A best-effort output limiter. It does not provide any fairness, and acts as a blocking gate-keeper
21  * for a sessions' channel.
22  */
23 @ThreadSafe
24 public final class ChannelOutputLimiter extends ChannelInboundHandlerAdapter {
25     private static final Logger LOG = LoggerFactory.getLogger(ChannelOutputLimiter.class);
26     private final BGPSessionImpl session;
27     private volatile boolean blocked;
28
29     ChannelOutputLimiter(final BGPSessionImpl session) {
30         this.session = Preconditions.checkNotNull(session);
31     }
32
33     private void ensureWritable() {
34         if (this.blocked) {
35             LOG.trace("Blocked slow path tripped on session {}", this.session);
36             synchronized (this) {
37                 while (this.blocked) {
38                     try {
39                         LOG.debug("Waiting for session {} to become writable", this.session);
40                         flush();
41                         this.wait();
42                     } catch (final InterruptedException e) {
43                         throw new IllegalStateException("Interrupted while waiting for channel to come back", e);
44                     }
45                 }
46
47                 LOG.debug("Resuming write on session {}", this.session);
48             }
49         }
50     }
51
52     public void write(final Notification msg) {
53         ensureWritable();
54         this.session.write(msg);
55     }
56
57     ChannelFuture writeAndFlush(final Notification msg) {
58         ensureWritable();
59         return this.session.writeAndFlush(msg);
60     }
61
62     public void flush() {
63         this.session.flush();
64     }
65
66     @Override
67     public void channelWritabilityChanged(final ChannelHandlerContext ctx) throws Exception {
68         final boolean w = ctx.channel().isWritable();
69
70         synchronized (this) {
71             this.blocked = !w;
72             LOG.debug("Writes on session {} {}", this.session, w ? "unblocked" : "blocked");
73
74             if (w) {
75                 notifyAll();
76             }
77         }
78
79         super.channelWritabilityChanged(ctx);
80     }
81
82     @Override
83     public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
84         synchronized (this) {
85             this.blocked = false;
86             notifyAll();
87         }
88
89         super.channelInactive(ctx);
90     }
91 }