Source code: mindbright/ssh/SSHPduQueue.java
1 /******************************************************************************
2 *
3 * Copyright (c) 1998,99 by Mindbright Technology AB, Stockholm, Sweden.
4 * www.mindbright.se, info@mindbright.se
5 *
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License as published by
8 * the Free Software Foundation; either version 2 of the License, or
9 * (at your option) any later version.
10 *
11 * This program is distributed in the hope that it will be useful,
12 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 * GNU General Public License for more details.
15 *
16 *****************************************************************************
17 * $Author: nallen $
18 * $Date: 2001/11/12 16:31:20 $
19 * $Name: $
20 *****************************************************************************/
21 package mindbright.ssh;
22
23 public final class SSHPduQueue {
24
25 final static int SSH_QUEUE_DEPTH = 512;
26 final static int SSH_QUEUE_HIWATER = 384;
27
28 Object[] queue;
29 Object queueLock;
30 boolean isWaitGet;
31 boolean isWaitPut;
32 int rOffset;
33 int wOffset;
34 int maxQueueDepth;
35
36 public SSHPduQueue() {
37 this.queue = new Object[SSH_QUEUE_DEPTH + 1];
38 this.queueLock = new Object();
39 this.isWaitGet = false;
40 this.isWaitPut = false;
41 this.rOffset = 0;
42 this.wOffset = 0;
43 this.maxQueueDepth = SSH_QUEUE_DEPTH;
44 }
45
46 public void setMaxDepth(int maxDepth) {
47 synchronized(queueLock) {
48 maxQueueDepth = maxDepth;
49 }
50 }
51
52 public void putLast(SSHPdu pdu) {
53 synchronized(queueLock) {
54 putFlowControl();
55 queue[wOffset++] = pdu;
56 if(wOffset == (SSH_QUEUE_DEPTH + 1))
57 wOffset = 0;
58 if(isWaitGet)
59 queueLock.notify();
60 }
61 }
62
63 public void putFirst(SSHPdu pdu) {
64 synchronized(queueLock) {
65 putFlowControl();
66 rOffset--;
67 if(rOffset == -1)
68 rOffset = SSH_QUEUE_DEPTH;
69 queue[rOffset] = pdu;
70 if(isWaitGet)
71 queueLock.notify();
72 }
73 }
74
75 public void release() {
76 synchronized(queueLock) {
77 if(isWaitGet)
78 queueLock.notify();
79 }
80 }
81
82 public boolean isEmpty() {
83 boolean isEmpty;
84 synchronized(queueLock) {
85 isEmpty = (rOffset == wOffset);
86 }
87 return isEmpty;
88 }
89
90 private final void putFlowControl() {
91 int fs = freeSpace();
92 if(fs == (SSH_QUEUE_DEPTH - maxQueueDepth)) {
93 isWaitPut = true;
94 }
95 if(isWaitPut) {
96 try {
97 queueLock.wait();
98 } catch (InterruptedException e) {
99 // !!!
100 }
101 }
102 }
103
104 private final int freeSpace() {
105 int fSpc = rOffset - wOffset;
106 if(fSpc <= 0)
107 fSpc += (SSH_QUEUE_DEPTH + 1);
108 fSpc--;
109 return fSpc;
110 }
111
112 public SSHPdu getFirst() {
113 SSHPdu pdu = null;
114
115 synchronized(queueLock) {
116 if(isEmpty()) {
117 isWaitGet = true;
118 try {
119 queueLock.wait();
120 } catch (InterruptedException e) {
121 // !!!
122 }
123 }
124 isWaitGet = false;
125 pdu = (SSHPdu) queue[rOffset];
126 queue[rOffset++] = null;
127 if(rOffset == (SSH_QUEUE_DEPTH + 1))
128 rOffset = 0;
129 if(isWaitPut && (freeSpace() > (SSH_QUEUE_DEPTH - SSH_QUEUE_HIWATER))) {
130 queueLock.notifyAll();
131 isWaitPut = false;
132 }
133 }
134
135 return pdu;
136 }
137
138 }