SlicingMeasurementConsumer.java

Violations

Source code

/*
  * hgcommons 9
 * Hammurapi Group Common Library
 * Copyright (C) 2003  Hammurapi Group
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2 of the License, or (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
 *
 * URL: http://www.hammurapi.biz/hammurapi-biz/ef/xmenu/hammurapi-group/products/products/hgcommons/index.html
 * e-Mail: support@hammurapi.biz
  */
package biz.hammurapi.metrics;

import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;

import biz.hammurapi.config.Component;
import biz.hammurapi.config.ConfigurationException;

/**
 * Slices metrics.
 * @author Pavel Vlasov
 * @version $Revision: 1.3 $
 */
public class SlicingMeasurementConsumer implements MeasurementConsumer, Component {
	
	private class SliceEntry {
		String category;
		Slice slice;
		/**
		 * @param category
		 * @param slice
		 */
		SliceEntry(String category, Slice slice) {
			super();
			this.category = category;
			this.slice = slice;
		}
	}
	
	private Map slices=new HashMap();
	private LinkedList sliceQueue=new LinkedList();
	private boolean keepMeasurements=false;
	private long tick=60000;
	private int maxQueue=1000;
	private SliceConsumer sliceConsumer=new ConsoleSliceConsumer();
	private Timer timer;
	private boolean isOwnTimer;
	
	protected SliceConsumer getSliceConsumer() {
		return sliceConsumer;
	}
	
	/**
	 * Creates a new instance with internal timer.
	 * @param tick Slice size in milliseconds
	 * @param keepMeasurements If true individual measurements are reported, only aggregated values otherwise
	 * @param maxQueue Maximum number of slices pending to be consumed. 0 - no limit. If sampling ratio is higher than
	 * consuming ration then excessive slices will be dropped with a notice on console.
	 */
	public SlicingMeasurementConsumer(long tick, boolean keepMeasurements, int maxQueue, SliceConsumer sliceConsumer) {
		this(tick, keepMeasurements, maxQueue, sliceConsumer, null);
	}
	
	/**
	 * Creates a new instance with internal timer.
	 * @param tick Slice size in milliseconds
	 * @param keepMeasurements If true individual measurements are reported, only aggregated values otherwise
	 * @param maxQueue Maximum number of slices pending to be consumed. 0 - no limit. If sampling ratio is higher than
	 * consuming ration then excessive slices will be dropped with a notice on console.
	 * @param timer Timer to use for slicing metrics and passing them to slice consumer. If it is null then an internal timer is
	 * created.
	 */
	public SlicingMeasurementConsumer(long tick, boolean keepMeasurements, int maxQueue, SliceConsumer sliceConsumer, Timer timer) {
		super();
		this.keepMeasurements = keepMeasurements;
		this.tick=tick;
		this.maxQueue=maxQueue;
		this.sliceConsumer=sliceConsumer;
		this.timer=timer;
	}
	
	/**
	 * Default constructor with default settings.
	 */
	public SlicingMeasurementConsumer() {
		super();
	}
		
	public void addMeasurement(String name, double value, long time) {
		synchronized (slices) {
			Slice slice=(Slice) slices.get(name);
			if (slice==null) {
				slice=new SimpleSlice(name, keepMeasurements);
				slices.put(name, slice);
			}
			slice.add(value, time);
			
			if (slice.getTo()-slice.getFrom()>=tick) {
				slices.remove(name);
				addSliceToQueue(null, slice);
			}
		}
	}
	
	private Map instances=new HashMap();
	
	private class CategorizedConsumer implements MeasurementConsumer, Component {
		
		
		String category;
		Map slices=new HashMap();
		
		/**
		 * @param category
		 */
		public CategorizedConsumer(String category) {
			super();
			this.category = category;
		}
		
		public void addMeasurement(String name, double value, long time) {
			synchronized (slices) {
				Slice slice=(Slice) slices.get(name);
				if (slice==null) {
					slice=new SimpleSlice(name, keepMeasurements);
					slices.put(name, slice);
				}
				slice.add(value, time);
				
				if (slice.getTo()-slice.getFrom()>=tick) {
					slices.remove(name);
					addSliceToQueue(category, slice);
				}
			}
		}

		public void start() throws ConfigurationException {
			SlicingMeasurementConsumer.this.start();
		}

		public void stop() throws ConfigurationException {
			SlicingMeasurementConsumer.this.stop();
		}

		public void setOwner(Object owner) {
			// Ignore		

		}	
	}
	
	/**
	 * @param category
	 * @return Instance for a category.
	 */
	public MeasurementConsumer getCategoryInstance(final String category) {
		synchronized (instances) {
			MeasurementConsumer ret = (MeasurementConsumer) instances.get(category);
			if (ret==null) {
				ret = new CategorizedConsumer(category);					
				instances.put(category, ret);
			}
			return ret;
		}
	}
	
	private int droppedCounter;
	private long firstDropped;
	private Thread slicingThread;
	
	public void shutdown() {
		to=System.currentTimeMillis();
        
        synchronized (slices) {
        	Iterator it=slices.values().iterator();
        	while (it.hasNext()) {
        		Slice slice=(Slice) it.next();
       			it.remove();
       			addSliceToQueue(null, slice);
        	}
        }
        
        synchronized (instances) {
        	Iterator iit=instances.values().iterator();
        	while (iit.hasNext()) {
        		CategorizedConsumer cConsumer=(CategorizedConsumer) iit.next();
        		synchronized (cConsumer.slices) {
                	Iterator it=cConsumer.slices.values().iterator();
                	while (it.hasNext()) {
                		Slice slice=(Slice) it.next();
               			it.remove();
               			addSliceToQueue(cConsumer.category, slice);
                	}
        		}
        	}
        }        		
		
//		MeasurementCategoryFactory.unregister(this); - Unregister explicitly!!!

	    addSliceToQueue(null,null);
	    try {
			slicingThread.join();
		} catch (InterruptedException e) {
			throw new MetricsException(e);
		}
		
		// Stop the time is it is our own.

		if (isOwnTimer) {
			timer.cancel();
		}
	}
	
	/**
	 *
	 */
	private void addSliceToQueue(String category, final Slice slice) {
		synchronized (sliceQueue) {
			if (slice!=null && maxQueue!=0 && sliceQueue.size()>maxQueue) {
			    firstDropped=slice.getTo();
				droppedCounter++;
			} else {
				if (droppedCounter>0) {
				    sliceQueue.add(
				    	new SliceEntry(
				    			"DroppedSlices",
								new Slice() {

		                        public long getFrom() {                            
		                            return firstDropped;
		                        }
		
		                        public long getTo() {
		                            return slice.getTo();
		                        }
		
		                        public int getNumber() {
		                            return droppedCounter;
		                        }
		
		                        public double getMin() {
		                            return 0;
		                        }
		
		                        public double getMax() {
		                            return 0;
		                        }
		
		                        public double getAvg() {
		                            return 0;
		                        }
		
		                        public double getTotal() {
		                            return 0;
		                        }
		
		                        public void add(double value, long time) {
		                            throw new UnsupportedOperationException();
		                        }
		
		                        public void add(Metric metric) {
		                            throw new UnsupportedOperationException();
		                        }
		
		                        public Collection getMeasurements() {
		                            return null;
		                        }
		
		                        public String getName() {
		                            return "DROPPED SLICES";
		                        }
		
		                        public double getDeviation() {
		                            return 0;
		                        }
						        
						    }));
					droppedCounter=0;
				}
				
				sliceQueue.add(slice==null ? null : new SliceEntry(category, slice));
				sliceQueue.notifyAll();
			}
		}
	}

	/**
     *
     */
    private void onTick() {
        long now=System.currentTimeMillis();
        
        synchronized (slices) {
        	Iterator it=slices.values().iterator();
        	while (it.hasNext()) {
        		Slice slice=(Slice) it.next();
        		if (now-slice.getFrom()>=tick) {
        			it.remove();
        			addSliceToQueue(null, slice);
        		}
        	}
        }
        
        synchronized (instances) {
        	Iterator iit=instances.values().iterator();
        	while (iit.hasNext()) {
        		CategorizedConsumer cConsumer=(CategorizedConsumer) iit.next();
        		synchronized (cConsumer.slices) {
                	Iterator it=cConsumer.slices.values().iterator();
                	while (it.hasNext()) {
                		Slice slice=(Slice) it.next();
                		if (now-slice.getFrom()>=tick) {
                			it.remove();
                			addSliceToQueue(cConsumer.category, slice);
                		}
                	}
        		}
        	}
        }
        
        if (sliceConsumer instanceof HousekeepingSliceConsumer) {
        	((HousekeepingSliceConsumer) sliceConsumer).onTick(now);
        }
    }
    
    {		
		slicingThread=new Thread() {
			{
				setName("Slice queue processor");
				setDaemon(true);
				setPriority(Thread.MIN_PRIORITY);
				start();
			}
			
			private SliceEntry getSliceEntry() throws InterruptedException {
				synchronized (sliceQueue) {
					while (sliceQueue.isEmpty()) {
						sliceQueue.wait();
					}
				
					return (SliceEntry) sliceQueue.removeFirst();					
				}
			}
			
			public void run() {
				while (true) {
					try {
						SliceEntry entry=getSliceEntry();
						if (entry==null) {
							if (sliceConsumer instanceof Component) {
								try {
									((Component) sliceConsumer).stop();
								} catch (ConfigurationException e1) {
									e1.printStackTrace();
								}
							}
							return;
						}
						
						// Put entry back to queue if slice consumer is unable to consume

						// and sleep.

						if (!sliceConsumer.consumeSlice(entry.category, entry.slice)) {
							synchronized (sliceQueue) {
								sliceQueue.add(entry);
							}
							sleep(tick);
						}
					} catch (InterruptedException e) {
						return;
					}
				}				
			}
		};		
	}

    private int useCounter;
    
    protected long from;
    protected long to;
    
    private TimerTask tickTask;
    
    /**
     * Increments use counter
     */
	public void start() throws ConfigurationException {
		if (timer==null) {
			timer=new Timer(true);
			isOwnTimer=true;
		}
		
		tickTask = new TimerTask() {
			public void run() {
				onTick();
			}
		};
		
		timer.scheduleAtFixedRate(tickTask, tick, tick);
		
		if (useCounter==0) {
			from=System.currentTimeMillis();
			if (sliceConsumer instanceof Component) {
				((Component) sliceConsumer).start();
			}
		}
		
		++useCounter;
	}

	/**
	 * Decrements use counter and invokes shutdown() when counter==0
	 */
	public void stop() throws ConfigurationException {
		if (--useCounter==0) {
			shutdown();
		}
	}    

	public void setOwner(Object owner) {
		// Ignore		

	}

	public int getMaxQueue() {
		return maxQueue;
	}

	public void setMaxQueue(int maxQueue) {
		this.maxQueue = maxQueue;
	}

	public long getTick() {
		return tick;
	}

	public void setTick(long tick) {
		this.tick = tick;
	}

	public void setSliceConsumer(SliceConsumer sliceConsumer) {
		this.sliceConsumer = sliceConsumer;
	}

	public boolean isKeepMeasurements() {
		return keepMeasurements;
	}

	public void setKeepMeasurements(boolean keepMeasurements) {
		this.keepMeasurements = keepMeasurements;
	}	
	
	
}