ListenerNotifier


1   // $Header: /home/cvs/jakarta-jmeter/src/core/org/apache/jmeter/threads/ListenerNotifier.java,v 1.16.2.1 2005/01/08 15:23:51 sebb Exp $
2   /*
3    * Copyright 2001-2004 The Apache Software Foundation.
4    *
5    * Licensed under the Apache License, Version 2.0 (the "License");
6    * you may not use this file except in compliance with the License.
7    * You may obtain a copy of the License at
8    *
9    *   http://www.apache.org/licenses/LICENSE-2.0
10   *
11   * Unless required by applicable law or agreed to in writing, software
12   * distributed under the License is distributed on an "AS IS" BASIS,
13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14   * See the License for the specific language governing permissions and
15   * limitations under the License.
16   * 
17  */
18  
19  package org.apache.jmeter.threads;
20  import java.util.Iterator  ;
21  import java.util.List  ;
22  
23  import org.apache.commons.collections.Buffer;
24  import org.apache.commons.collections.BufferUtils;
25  import org.apache.commons.collections.UnboundedFifoBuffer;
26  import org.apache.jmeter.samplers.SampleEvent;
27  import org.apache.jmeter.samplers.SampleListener;
28  import org.apache.jorphan.logging.LoggingManager;
29  import org.apache.log.Logger;
30  
31  /**
32   * The <code>ListenerNotifier</code> thread is responsible for performing
33   * asynchronous notifications that a sample has occurred.  Each time a
34   * sample occurs, the <code>addLast</code> method should be called to add
35   * the sample and its list of listeners to the notification queue.  This
36   * thread will then notify those listeners asynchronously at some future
37   * time.
38   * <p>
39   * In the current implementation, the notifications will be made in batches,
40   * with 2 seconds between the beginning of successive batches.  If the
41   * notifier thread starts to get behind, the priority of the thread will be
42   * increased in an attempt to help it to keep up.
43   *
44   * @see org.apache.jmeter.samplers.SampleListener
45   * 
46   * @version $Revision: 1.16.2.1 $
47   */
48  public class ListenerNotifier
49  {
50      private static Logger log = LoggingManager.getLoggerForClass();
51  
52      /**
53       * The number of milliseconds between batches of notifications.
54       */
55      private static final int SLEEP_TIME = 2000;
56  
57      /**
58       * Indicates whether or not this thread should remain running.  The
59       * thread will continue running after this field is set to false until
60       * the next batch of notifications has been completed and the
61       * notification queue is empty.
62       */
63      private boolean running = true;
64  
65      /**
66       * Indicates whether or not this thread has stopped.  No further
67       * notifications will be performed.
68       */
69      private boolean isStopped = true;
70  
71      /**
72       * The queue containing the notifications to be performed.  Each
73       * notification consists of a pair of entries in this queue.  The
74       * first is the {@link org.apache.jmeter.samplers.SampleEvent
75       * SampleEvent} representing the sample.  The second is a List of
76       * {@link org.apache.jmeter.samplers.SampleListener SampleListener}s
77       * which should be notified.
78       */
79      private Buffer listenerEvents =
80          BufferUtils.synchronizedBuffer(new UnboundedFifoBuffer());
81  
82  
83  
84      /**
85       * Stops the ListenerNotifier thread.  The thread will continue processing
86       * any events remaining in the notification queue before it actually
87       * stops, but this method will return immediately.
88       */
89      public void stop()
90      {
91          running = false;
92      }
93  
94      /**
95       * Indicates whether or not the thread has stopped.  This will not
96       * return true until the <code>stop</code> method has been called and
97       * any remaining notifications in the queue have been completed.
98       *
99       * @return  true if the ListenerNotifier has completely stopped, false
100      *          otherwise
101      */
102     public boolean isStopped()
103     {
104         return isStopped;
105     }
106 
107     /**
108      * Process the events in the notification queue until the thread has
109      * been told to stop and the notification queue is empty.
110      * <p>
111      * In the current implementation, this method will iterate continually
112      * until the thread is told to stop.  In each iteration it will process
113      * any notifications that are in the queue at the beginning of the
114      * iteration, and then will sleep until it is time to start the next
115      * batch.  As long as the thread is keeping up, each batch should start
116      * 2 seconds after the beginning of the last batch.  This exact
117      * behavior is subject to change.
118      */
119     public void run()
120     {
121         boolean isMaximumPriority = false;
122         int normalCount = 0;
123 
124         while (running)
125         {
126             long startTime = System.currentTimeMillis();
127             processNotifications();
128             long sleep = SLEEP_TIME - (System.currentTimeMillis() - startTime);
129 
130             // If the thread has been told to stop then we shouldn't sleep
131             if (!running)
132             {
133                 break;
134             }
135 
136             if (sleep < 0)
137             {
138                 isMaximumPriority = true;
139                 normalCount = 0;
140                 if (log.isInfoEnabled())
141                 {
142                     log.info("ListenerNotifier exceeded maximum " +
143                             "notification time by " + (-sleep) + "ms");
144                 }
145                 boostPriority();
146             }
147             else
148             {
149                 normalCount++;
150 
151                 // If there have been three consecutive iterations since the
152                 // last iteration which took too long to execute, return the
153                 // thread to normal priority.
154                 if (isMaximumPriority && normalCount >= 3)
155                 {
156                     isMaximumPriority = false;
157                     unboostPriority();
158                 }
159 
160                 if (log.isDebugEnabled())
161                 {
162                     log.debug("ListenerNotifier sleeping for " + sleep + "ms");
163                 }
164 
165                 try
166                 {
167                     Thread.sleep(sleep);
168                 }
169                 catch (InterruptedException   e)
170                 {
171                 }
172             }
173         }
174 
175         // Make sure that all pending notifications are processed before
176         // actually ending the thread.
177         processNotifications();
178         isStopped = true;
179     }
180 
181     /**
182      * Process all of the pending notifications.  Only the samples which are
183      * in the queue when this method is called will be processed.  Any samples
184      * added between the time when this method is called and when it exits are
185      * saved for the next batch.
186      */
187     private void processNotifications()
188     {
189         int listenerEventsSize = listenerEvents.size();
190         if (log.isDebugEnabled())
191         {
192             log.debug ("ListenerNotifier: processing " + listenerEventsSize +
193                     " events");
194         }
195 
196         while (listenerEventsSize > 0)
197         {
198             // Since this is a FIFO and this is the only place we remove
199             // from it (only from a single thread) we don't have to remove
200             // these two items in one atomic operation.  Each individual
201             // remove is atomic (because we use a synchronized buffer),
202             // which is necessary since the buffer can be accessed from
203             // other threads (to add things to the buffer).
204             SampleEvent res = (SampleEvent)listenerEvents.remove();
205             List   listeners = (List  )listenerEvents.remove();
206 
207             notifyListeners (res, listeners);
208 
209             listenerEventsSize -= 2;
210         }
211     }
212 
213     /**
214      * Boost the priority of the current thread to maximum priority.  If
215      * the thread is already at maximum priority then this will have no
216      * effect.
217      */
218     private void boostPriority()
219     {
220         if (Thread.currentThread().getPriority() != Thread.MAX_PRIORITY)
221         {
222             log.info("ListenerNotifier: Boosting thread priority to maximum.");
223             Thread.currentThread().setPriority(Thread.MAX_PRIORITY);
224         }
225     }
226 
227     /**
228      * Return the priority of the current thread to normal.  If the thread
229      * is already at normal priority then this will have no effect.
230      */
231     private void unboostPriority()
232     {
233         if (Thread.currentThread().getPriority() != Thread.NORM_PRIORITY)
234         {
235             log.info("ListenerNotifier: Returning thread priority to normal.");
236             Thread.currentThread().setPriority(Thread.NORM_PRIORITY);
237         }
238     }
239 
240     /**
241      * Notify a list of listeners that a sample has occurred.
242      *
243      * @param res       the sample event that has occurred.  Must be non-null.
244      * @param listeners a list of the listeners which should be notified.
245      *                  This list must not be null and must contain only
246      *                  SampleListener elements.
247      */
248     public void notifyListeners(SampleEvent res, List   listeners)
249     {
250         Iterator   iter = listeners.iterator();
251         while (iter.hasNext())
252         {
253             try {
254                 ((SampleListener) iter.next()).sampleOccurred(res);
255             } catch (RuntimeException   e) {
256                 log.error("Detected problem in Listener: ",e);
257                 log.info("Continuing to process further listeners");
258             }
259         }
260     }
261 
262     /**
263      * Add a new sample event to the notification queue.  The notification
264      * will be performed asynchronously and this method will return
265      * immediately.
266      *
267      * @param item      the sample event that has occurred.  Must be non-null.
268      * @param listeners a list of the listeners which should be notified.
269      *                  This list must not be null and must contain only
270      *                  SampleListener elements.
271      */
272     public void addLast(SampleEvent item, List   listeners)
273     {
274         // Must use explicit synchronization here so that the item and
275         // listeners are added together atomically
276         synchronized (listenerEvents)
277         {
278             listenerEvents.add(item);
279             listenerEvents.add(listeners);
280         }
281     }
282 }
283
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags