content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  
  package org.apache.hadoop.chukwa.datacollection.agent;
  
  
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Queue;
  import org.apache.hadoop.chukwa.Chunk;
  import org.apache.hadoop.chukwa.datacollection.ChunkQueue;
  import org.apache.hadoop.chukwa.datacollection.agent.metrics.ChunkQueueMetrics;
  import org.apache.hadoop.conf.Configuration;
  import org.apache.log4j.Logger;
  
  /**
   * An event queue that blocks once a fixed upper limit of data is enqueued.
   * 
   * For now, uses the size of the data field. Should really use
   * estimatedSerializedSize()?
   * 
   */
  public class MemLimitQueue implements ChunkQueue {
    static Logger log = Logger.getLogger(WaitingQueue.class);
    static final ChunkQueueMetrics metrics = new ChunkQueueMetrics("chukwaAgent", "chunkQueue");;
    private Queue<Chunk> queue = new LinkedList<Chunk>();
    private long dataSize = 0;
    private long MAX_MEM_USAGE;
    static final String CHUNK_QUEUE_LIMIT = "chukwaAgent.chunk.queue.limit";
    static final int QUEUE_SIZE = 10 * 1024 * 1024;
  
    public MemLimitQueue(Configuration conf) {
      configure(conf);
    }
  
    /**
     * @see org.apache.hadoop.chukwa.datacollection.ChunkQueue#add(org.apache.hadoop.chukwa.Chunk)
     */
    public void add(Chunk chunk) throws InterruptedException {
      assert chunk != null : "can't enqueue null chunks";
      synchronized (this) {
        while (chunk.getData().length + dataSize > MAX_MEM_USAGE) {
          try {
            if(dataSize == 0) { //queue is empty, but data is still too big
              log.error("JUMBO CHUNK SPOTTED: type= " + chunk.getDataType() + 
                  " and source =" +chunk.getStreamName()); 
              return; //return without sending; otherwise we'd deadlock.
              //this error should probably be fatal; there's no way to recover.
            }
            metrics.fullQueue.set(1);
            this.wait();
            log.info("MemLimitQueue is full [" + dataSize + "]");
          } catch (InterruptedException e) {
          }
        }
        metrics.fullQueue.set(0);
        dataSize += chunk.getData().length;
        queue.add(chunk);
        metrics.addedChunk.inc();
        metrics.queueSize.set(queue.size());
        metrics.dataSize.set(dataSize);
        this.notifyAll();
      }
  
    }
  
    /**
     * @see org.apache.hadoop.chukwa.datacollection.ChunkQueue#collect(java.util.List,
     *      int)
     */
    public void collect(List<Chunk> events, int maxSize)
        throws InterruptedException {
      synchronized (this) {
        // we can't just say queue.take() here, since we're holding a lock.
        while (queue.isEmpty()) {
          this.wait();
        }
  
        int size = 0;
        while (!queue.isEmpty() && (size < maxSize)) {
          Chunk e = this.queue.remove();
          metrics.removedChunk.inc();
          int chunkSize = e.getData().length;
          size += chunkSize;
          dataSize -= chunkSize;
          metrics.dataSize.set(dataSize);
          events.add(e);
        }
        metrics.queueSize.set(queue.size());
        this.notifyAll();
      }
  
      if (log.isDebugEnabled()) {
        log.debug("WaitingQueue.inQueueCount:" + queue.size()
            + "\tWaitingQueue.collectCount:" + events.size());
      }
    }
  
    public int size() {
      return queue.size();
    }
    
    private void configure(Configuration conf) {
      MAX_MEM_USAGE = QUEUE_SIZE;
      if(conf == null){
        return;
      }
      String limit = conf.get(CHUNK_QUEUE_LIMIT);
      if(limit != null){
        try{
          MAX_MEM_USAGE = Integer.parseInt(limit);
        } catch(NumberFormatException nfe) {
          log.error("Exception reading property " + CHUNK_QUEUE_LIMIT
              + ". Defaulting internal queue size to " + QUEUE_SIZE);
        }
      }
      log.info("Using MemLimitQueue limit of " + MAX_MEM_USAGE);
    }
  }
  