Fix deadlock in AsyncSshHandlerWriter
[controller.git] / opendaylight / netconf / netconf-netty-util / src / main / java / org / opendaylight / controller / netconf / nettyutil / handler / ssh / client / AsyncSshHandlerWriter.java
1 /*
2  * Copyright (c) 2014 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
9 package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client;
10
11 import com.google.common.base.Charsets;
12 import com.google.common.base.Preconditions;
13 import io.netty.buffer.ByteBuf;
14 import io.netty.channel.ChannelHandlerContext;
15 import io.netty.channel.ChannelPromise;
16 import java.util.Deque;
17 import java.util.LinkedList;
18 import java.util.Queue;
19 import org.apache.sshd.common.future.SshFutureListener;
20 import org.apache.sshd.common.io.IoOutputStream;
21 import org.apache.sshd.common.io.IoWriteFuture;
22 import org.apache.sshd.common.io.WritePendingException;
23 import org.apache.sshd.common.util.Buffer;
24 import org.slf4j.Logger;
25 import org.slf4j.LoggerFactory;
26
27 /**
28  * Async Ssh writer. Takes messages(byte arrays) and sends them encrypted to remote server.
29  * Also handles pending writes by caching requests until pending state is over.
30  */
31 public final class AsyncSshHandlerWriter implements AutoCloseable {
32
33     private static final Logger LOG = LoggerFactory
34             .getLogger(AsyncSshHandlerWriter.class);
35
36     // public static final int MAX_PENDING_WRITES = 1000;
37     // TODO implement Limiting mechanism for pending writes
38     // But there is a possible issue with limiting:
39     // 1. What to do when queue is full ? Immediate Fail for every request ?
40     // 2. At this level we might be dealing with Chunks of messages(not whole messages) and unexpected behavior might occur
41     // when we send/queue 1 chunk and fail the other chunks
42
43     private volatile IoOutputStream asyncIn;
44
45     // Order has to be preserved for queued writes
46     private final Deque<PendingWriteRequest> pending = new LinkedList<>();
47
48     public AsyncSshHandlerWriter(final IoOutputStream asyncIn) {
49         this.asyncIn = asyncIn;
50     }
51
52     public void write(final ChannelHandlerContext ctx,
53             final Object msg, final ChannelPromise promise) {
54         if (asyncIn == null) {
55             promise.setFailure(new IllegalStateException("Channel closed"));
56             return;
57         }
58         // synchronized block due to deadlock that happens on ssh window resize
59         // writes and pending writes would lock the underlyinch channel session
60         // window resize write would try to write the message on an already locked channelSession
61         // while the pending write was in progress from the write callback
62         synchronized (asyncIn) {
63             // TODO check for isClosed, isClosing might be performed by mina SSH internally and is not required here
64             // If we are closed/closing, set immediate fail
65             if (asyncIn.isClosed() || asyncIn.isClosing()) {
66                 promise.setFailure(new IllegalStateException("Channel closed"));
67             } else {
68                 final ByteBuf byteBufMsg = (ByteBuf) msg;
69                 if (pending.isEmpty() == false) {
70                     queueRequest(ctx, byteBufMsg, promise);
71                     return;
72                 }
73
74                 writeWithPendingDetection(ctx, promise, byteBufMsg, false);
75             }
76         }
77     }
78
79     //sending message with pending
80     //if resending message not succesfull, then attribute wasPending is true
81     private void writeWithPendingDetection(final ChannelHandlerContext ctx, final ChannelPromise promise, final ByteBuf byteBufMsg, final boolean wasPending) {
82         try {
83
84             if (LOG.isTraceEnabled()) {
85                 LOG.trace("Writing request on channel: {}, message: {}", ctx.channel(), byteBufToString(byteBufMsg));
86             }
87             asyncIn.write(toBuffer(byteBufMsg)).addListener(new SshFutureListener<IoWriteFuture>() {
88
89                 @Override
90                 public void operationComplete(final IoWriteFuture future) {
91                     // synchronized block due to deadlock that happens on ssh window resize
92                     // writes and pending writes would lock the underlyinch channel session
93                     // window resize write would try to write the message on an already locked channelSession,
94                     // while the pending write was in progress from the write callback
95                     synchronized (asyncIn) {
96                         if (LOG.isTraceEnabled()) {
97                             LOG.trace("Ssh write request finished on channel: {} with result: {}: and ex:{}, message: {}",
98                                     ctx.channel(), future.isWritten(), future.getException(), byteBufToString(byteBufMsg));
99                         }
100
101                         // Notify success or failure
102                         if (future.isWritten()) {
103                             promise.setSuccess();
104                         } else {
105                             LOG.warn("Ssh write request failed on channel: {} for message: {}", ctx.channel(), byteBufToString(byteBufMsg), future.getException());
106                             promise.setFailure(future.getException());
107                         }
108
109                         // Not needed anymore, release
110                         byteBufMsg.release();
111
112                         //rescheduling message from queue after successfully sent
113                         if (wasPending) {
114                             byteBufMsg.resetReaderIndex();
115                             pending.remove();
116                         }
117                     }
118
119                     // Check pending queue and schedule next
120                     // At this time we are guaranteed that we are not in pending state anymore so the next request should succeed
121                     writePendingIfAny();
122                 }
123             });
124
125         } catch (final WritePendingException e) {
126
127             if(wasPending == false){
128                 queueRequest(ctx, byteBufMsg, promise);
129             }
130         }
131     }
132
133     private void writePendingIfAny() {
134         synchronized (asyncIn) {
135             if (pending.peek() == null) {
136                 return;
137             }
138
139             final PendingWriteRequest pendingWrite = pending.peek();
140             final ByteBuf msg = pendingWrite.msg;
141             if (LOG.isTraceEnabled()) {
142                 LOG.trace("Writing pending request on channel: {}, message: {}", pendingWrite.ctx.channel(), byteBufToString(msg));
143             }
144
145             writeWithPendingDetection(pendingWrite.ctx, pendingWrite.promise, msg, true);
146         }
147     }
148
149     public static String byteBufToString(final ByteBuf msg) {
150         final String s = msg.toString(Charsets.UTF_8);
151         msg.resetReaderIndex();
152         return s;
153     }
154
155     private void queueRequest(final ChannelHandlerContext ctx, final ByteBuf msg, final ChannelPromise promise) {
156 //        try {
157         LOG.debug("Write pending on channel: {}, queueing, current queue size: {}", ctx.channel(), pending.size());
158         if (LOG.isTraceEnabled()) {
159             LOG.trace("Queueing request due to pending: {}", byteBufToString(msg));
160         }
161         new PendingWriteRequest(ctx, msg, promise).pend(pending);
162 //        } catch (final Exception ex) {
163 //            LOG.warn("Unable to queue write request on channel: {}. Setting fail for the request: {}", ctx.channel(), ex, byteBufToString(msg));
164 //            msg.release();
165 //            promise.setFailure(ex);
166 //        }
167     }
168
169     @Override
170     public void close() {
171         asyncIn = null;
172     }
173
174     private Buffer toBuffer(final ByteBuf msg) {
175         // TODO Buffer vs ByteBuf translate, Can we handle that better ?
176         msg.resetReaderIndex();
177         final byte[] temp = new byte[msg.readableBytes()];
178         msg.readBytes(temp, 0, msg.readableBytes());
179         return new Buffer(temp);
180     }
181
182     private static final class PendingWriteRequest {
183         private final ChannelHandlerContext ctx;
184         private final ByteBuf msg;
185         private final ChannelPromise promise;
186
187         public PendingWriteRequest(final ChannelHandlerContext ctx, final ByteBuf msg, final ChannelPromise promise) {
188             this.ctx = ctx;
189             // Reset reader index, last write (failed) attempt moved index to the end
190             msg.resetReaderIndex();
191             this.msg = msg;
192             this.promise = promise;
193         }
194
195         public void pend(final Queue<PendingWriteRequest> pending) {
196             // Preconditions.checkState(pending.size() < MAX_PENDING_WRITES,
197             // "Too much pending writes(%s) on channel: %s, remote window is not getting read or is too small",
198             // pending.size(), ctx.channel());
199             Preconditions.checkState(pending.offer(this), "Cannot pend another request write (pending count: %s) on channel: %s",
200                     pending.size(), ctx.channel());
201         }
202     }
203 }

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.