/*
 * 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 discards incoming chunks once a fixed upper limit of data
 * is enqueued. The method calling add will not block.
 * 
 * For now, uses the size of the data field. Should really use
 * estimatedSerializedSize()?
 * 
 */
public class NonBlockingMemLimitQueue implements ChunkQueue {
  static Logger log = Logger.getLogger(NonBlockingMemLimitQueue.class);
  static final ChunkQueueMetrics metrics = new ChunkQueueMetrics("chukwaAgent",
      "chunkQueue");
  static final String CHUNK_QUEUE_LIMIT = "chukwaAgent.chunk.queue.limit";
  static final int QUEUE_SIZE = 10 * 1024 * 1024;
  private Queue<Chunk> queue = new LinkedList<Chunk>();
  private long dataSize = 0;
  private long MAX_MEM_USAGE;

  public NonBlockingMemLimitQueue(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";
    int chunkSize = chunk.getData().length;
    synchronized (this) {
      if (chunkSize + dataSize > MAX_MEM_USAGE) {
        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.
        } else {
          metrics.fullQueue.set(1);
          log.warn("Discarding chunk due to NonBlockingMemLimitQueue full [" + dataSize
              + "]");
          return;
        }
      }
      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 NonBlockingMemLimitQueue limit of " + MAX_MEM_USAGE);
  }
}