KickJava   Java API By Example, From Geeks To Geeks.

Java > Open Source Codes > com > sun > jmx > snmp > daemon > SnmpQManager


1 /*
2  * @(#)file SnmpQManager.java
3  * @(#)author Sun Microsystems, Inc.
4  * @(#)version 1.6
5  * @(#)date 08/02/09
6  *
7  * Copyright 2004 Sun Microsystems, Inc. All rights reserved.
8  * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
9  *
10  */

11
12 package com.sun.jmx.snmp.daemon;
13
14 import java.util.Vector JavaDoc;
15 import java.io.Serializable JavaDoc;
16
17 // import debug stuff
18
//
19
import com.sun.jmx.trace.Trace;
20
21 /**
22  * This class implements a server queue manager.
23  * This class is for internal use.
24  */

25
26 final class SnmpQManager implements Serializable JavaDoc {
27
28     // VARIABLES
29
//----------
30

31     private SendQ newq ;
32     private WaitQ waitq ;
33
34     private ThreadGroup JavaDoc queueThreadGroup = null ;
35     private Thread JavaDoc requestQThread = null ;
36     private Thread JavaDoc timerQThread = null ;
37
38     static String JavaDoc dbgTag = "SnmpQManager";
39     
40     
41     // CONSTRUCTORS
42
//-------------
43

44     SnmpQManager() {
45         newq = new SendQ(20, 5) ;
46         waitq = new WaitQ(20, 5) ;
47
48         queueThreadGroup = new ThreadGroup JavaDoc("Qmanager Thread Group") ;
49     
50         // TIME BOMB HERE
51
startQThreads() ;
52     }
53     
54     public void startQThreads() {
55         if (timerQThread == null || timerQThread.isAlive() == false) {
56             timerQThread = new SnmpTimerServer(queueThreadGroup, this) ;
57         }
58         if (requestQThread == null || requestQThread.isAlive() == false) {
59             requestQThread = new SnmpSendServer(queueThreadGroup, this) ;
60         }
61     }
62
63     public void stopQThreads() {
64         
65         ((SnmpTimerServer)timerQThread).isBeingDestroyed = true;
66         waitq.isBeingDestroyed = true;
67         ((SnmpSendServer)requestQThread).isBeingDestroyed = true;
68         newq.isBeingDestroyed = true;
69         
70         if (timerQThread != null && timerQThread.isAlive() == true) {
71             ((SnmpTimerServer)timerQThread).stopTimerServer();
72         }
73         waitq = null;
74         timerQThread = null;
75         
76         if (requestQThread != null && requestQThread.isAlive() == true) {
77             ((SnmpSendServer)requestQThread).stopSendServer();
78         }
79         newq = null;
80         requestQThread = null;
81     }
82     
83     public void addRequest(SnmpInformRequest reqc) {
84         newq.addRequest(reqc) ;
85         return ;
86     }
87
88     public void addWaiting(SnmpInformRequest reqc) {
89         waitq.addWaiting(reqc) ;
90         return ;
91     }
92
93     public Vector JavaDoc getAllOutstandingRequest(long range) {
94         return newq.getAllOutstandingRequest(range) ;
95     }
96
97     public SnmpInformRequest getTimeoutRequests() {
98         return waitq.getTimeoutRequests() ;
99     }
100
101     public void removeRequest(SnmpInformRequest reqc) {
102         newq.removeElement(reqc) ;
103         waitq.removeElement(reqc) ;
104     }
105
106     public SnmpInformRequest removeRequest(long reqid) {
107         SnmpInformRequest reqc = null ;
108
109         if ((reqc = newq.removeRequest(reqid)) == null)
110             reqc = waitq.removeRequest(reqid) ;
111     
112         return reqc ;
113     }
114
115     // TRACES & DEBUG
116
//---------------
117

118     static boolean isTraceOn() {
119         return Trace.isSelected(Trace.LEVEL_TRACE, Trace.INFO_ADAPTOR_SNMP);
120     }
121
122     static void trace(String JavaDoc clz, String JavaDoc func, String JavaDoc info) {
123         Trace.send(Trace.LEVEL_TRACE, Trace.INFO_ADAPTOR_SNMP, clz, func, info);
124     }
125
126     static void trace(String JavaDoc func, String JavaDoc info) {
127         SnmpQManager.trace(dbgTag, func, info);
128     }
129     
130     static boolean isDebugOn() {
131         return Trace.isSelected(Trace.LEVEL_DEBUG, Trace.INFO_ADAPTOR_SNMP);
132     }
133
134     static void debug(String JavaDoc clz, String JavaDoc func, String JavaDoc info) {
135         Trace.send(Trace.LEVEL_DEBUG, Trace.INFO_ADAPTOR_SNMP, clz, func, info);
136     }
137
138     static void debug(String JavaDoc clz, String JavaDoc func, Throwable JavaDoc exception) {
139         Trace.send(Trace.LEVEL_DEBUG, Trace.INFO_ADAPTOR_SNMP, clz, func, exception);
140     }
141     
142     static void debug(String JavaDoc func, String JavaDoc info) {
143         SnmpQManager.debug(dbgTag, func, info);
144     }
145     
146     static void debug(String JavaDoc func, Throwable JavaDoc exception) {
147         SnmpQManager.debug(dbgTag, func, exception);
148     }
149     
150 }
151
152 /**
153  * This vector manages the inform requests to be sent to the manager.
154  */

155 class SendQ extends Vector JavaDoc {
156     
157     SendQ(int initialCapacity, int capacityIncr) {
158         super(initialCapacity , capacityIncr) ;
159     }
160
161     private synchronized void notifyClients() {
162         this.notifyAll() ;
163     }
164
165     public synchronized void addRequest(SnmpInformRequest req) {
166                 
167         long nextPoll = req.getAbsNextPollTime() ;
168
169         int i ;
170         for (i = size() ; i > 0 ; i--) {
171             if (nextPoll < getRequestAt(i-1).getAbsNextPollTime())
172                 break ;
173         }
174         if (i == size()) {
175             addElement(req) ;
176             notifyClients() ;
177         } else
178             insertElementAt(req, i) ;
179         return ;
180     }
181
182     public synchronized boolean waitUntilReady() {
183         while (true) {
184             if (isBeingDestroyed == true)
185                 return false;
186             long tmp = 0 ;
187             if (isEmpty() == false) {
188                 long currTime = System.currentTimeMillis() ;
189                 SnmpInformRequest req = (SnmpInformRequest) lastElement() ;
190                 tmp = req.getAbsNextPollTime() - currTime ;
191                 if (tmp <= 0) {
192                     return true ;
193                 }
194             }
195             waitOnThisQueue(tmp) ;
196         }
197     }
198
199     public synchronized Vector JavaDoc getAllOutstandingRequest(long margin) {
200         int i ;
201         Vector JavaDoc outreq = new Vector JavaDoc() ;
202         while (true) {
203             if (waitUntilReady() == true) {
204                 long refTime = System.currentTimeMillis() + margin ;
205
206                 for (i = size() ; i > 0 ; i--) {
207                     SnmpInformRequest req = getRequestAt(i-1) ;
208                     if (req.getAbsNextPollTime() > refTime)
209                         break ;
210                     outreq.addElement(req) ;
211                 }
212             
213                 if (! outreq.isEmpty()) {
214                     elementCount -= outreq.size() ;
215                     return outreq ;
216                 }
217             }
218             else
219                 return null;
220         }
221     }
222
223     public synchronized void waitOnThisQueue(long time) {
224         if (time == 0 && !isEmpty()) {
225             if (SnmpQManager.isDebugOn()) {
226                 SnmpQManager.debug("waitOnThisQueue", "[" + Thread.currentThread().toString() + "]:" +
227                                    "Fatal BUG :: Blocking on newq permenantly. But size = " + size());
228             }
229         }
230
231         try {
232             this.wait(time) ;
233         } catch (InterruptedException JavaDoc e) {
234         }
235     }
236
237     public SnmpInformRequest getRequestAt(int idx) {
238         return (SnmpInformRequest)elementAt(idx) ;
239     }
240     
241     public synchronized SnmpInformRequest removeRequest(long reqid) {
242         int max= size() ;
243         for (int i = 0 ; i < max ; i++) {
244             SnmpInformRequest reqc = getRequestAt(i) ;
245             if (reqid == reqc.getRequestId()) {
246                 removeElementAt(i) ;
247                 return reqc ;
248             }
249         }
250         return null ;
251     }
252
253     // This boolean is used to stop handling requests while the corresponding SnmpQManager
254
// is being destroyed.
255
//
256
boolean isBeingDestroyed = false;
257 }
258
259 /**
260  * This vector manages the inform requests to be retried to the manager.
261  */

262 class WaitQ extends Vector JavaDoc {
263     
264     WaitQ(int initialCapacity, int capacityIncr) {
265         super(initialCapacity , capacityIncr) ;
266     }
267
268     public synchronized void addWaiting(SnmpInformRequest req) {
269         
270         long waitTime = req.getAbsMaxTimeToWait() ;
271         int i ;
272         for (i = size() ; i > 0 ; i--) {
273             if (waitTime < getRequestAt(i-1).getAbsMaxTimeToWait())
274                 break ;
275         }
276         if (i == size()) {
277             addElement(req) ;
278             notifyClients() ;
279         } else
280             insertElementAt(req, i) ;
281         return ;
282     }
283
284     public synchronized boolean waitUntilReady() {
285         while (true) {
286             if (isBeingDestroyed == true)
287                 return false;
288             long tmp = 0 ;
289             if (isEmpty() == false) {
290                 long currTime = System.currentTimeMillis() ;
291                 SnmpInformRequest req = (SnmpInformRequest) lastElement() ;
292                 tmp = req.getAbsMaxTimeToWait() - currTime ;
293                 if (tmp <= 0) {
294                     return true ;
295                 }
296             }
297             waitOnThisQueue(tmp) ;
298         }
299     }
300
301     public synchronized SnmpInformRequest getTimeoutRequests() {
302         if (waitUntilReady() == true) {
303             SnmpInformRequest req = (SnmpInformRequest) lastElement() ;
304             elementCount-- ;
305             return req ;
306         }
307         else {
308             return null;
309         }
310     }
311
312     private synchronized void notifyClients() {
313         this.notifyAll() ;
314     }
315
316     public synchronized void waitOnThisQueue(long time) {
317         if (time == 0 && !isEmpty()) {
318             if (SnmpQManager.isDebugOn()) {
319                 SnmpQManager.debug("waitOnThisQueue", "[" + Thread.currentThread().toString() + "]:" +
320                                    "Fatal BUG :: Blocking on waitq permenantly. But size = " + size());
321             }
322         }
323                 
324         try {
325             this.wait(time) ;
326         } catch (InterruptedException JavaDoc e) {
327         }
328     }
329
330     public SnmpInformRequest getRequestAt(int idx) {
331         return (SnmpInformRequest)elementAt(idx) ;
332     }
333
334     public synchronized SnmpInformRequest removeRequest(long reqid) {
335         int max= size();
336         for (int i = 0 ; i < max ; i++) {
337             SnmpInformRequest reqc = getRequestAt(i) ;
338             if (reqid == reqc.getRequestId()) {
339                 removeElementAt(i) ;
340                 return reqc ;
341             }
342         }
343         return null ;
344     }
345
346     // This boolean is used to stop handling requests while the corresponding SnmpQManager
347
// is being destroyed.
348
//
349
boolean isBeingDestroyed = false;
350 }
351
Popular Tags