/**
 * 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 com.datatorrent.stram.engine;

import java.beans.IntrospectionException;
import java.beans.Introspector;
import java.beans.PropertyDescriptor;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import java.lang.reflect.Array;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.apex.common.util.AsyncStorageAgent;
import org.apache.hadoop.util.ReflectionUtils;

import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.math.IntMath;

import com.datatorrent.api.AutoMetric;
import com.datatorrent.api.Component;
import com.datatorrent.api.Context;
import com.datatorrent.api.InputOperator;
import com.datatorrent.api.Operator;
import com.datatorrent.api.Operator.InputPort;
import com.datatorrent.api.Operator.OutputPort;
import com.datatorrent.api.Operator.ProcessingMode;
import com.datatorrent.api.Operator.Unifier;
import com.datatorrent.api.Sink;
import com.datatorrent.api.Stats;
import com.datatorrent.api.StatsListener;
import com.datatorrent.api.StatsListener.OperatorRequest;
import com.datatorrent.api.StorageAgent;
import com.datatorrent.bufferserver.util.Codec;
import com.datatorrent.common.util.Pair;
import com.datatorrent.stram.api.Checkpoint;
import com.datatorrent.stram.api.OperatorDeployInfo;
import com.datatorrent.stram.api.StreamingContainerUmbilicalProtocol.ContainerStats;
import com.datatorrent.stram.debug.MuxSink;
import com.datatorrent.stram.plan.logical.Operators;
import com.datatorrent.stram.plan.logical.Operators.PortContextPair;
import com.datatorrent.stram.plan.logical.Operators.PortMappingDescriptor;
import com.datatorrent.stram.tuple.EndStreamTuple;
import com.datatorrent.stram.tuple.EndWindowTuple;

/**
 * <p>
 * Abstract Node class.</p>
 *
 * @param <OPERATOR>
 * @since 0.3.2
 */
public abstract class Node<OPERATOR extends Operator> implements Component<OperatorContext>, Runnable
{
  /**
   * if the Component is capable of taking only 1 input, call it INPUT.
   */
  public static final String INPUT = "input";
  /**
   * if the Component is capable of providing only 1 output, call it OUTPUT.
   */
  public static final String OUTPUT = "output";
  protected int APPLICATION_WINDOW_COUNT; /* this is write once variable */

  protected int DAG_CHECKPOINT_WINDOW_COUNT; /* this is write once variable */

  protected int CHECKPOINT_WINDOW_COUNT; /* this is write once variable */

  protected boolean DATA_TUPLE_AWARE; /* this is write once variable */

  protected int id;
  protected final HashMap<String, Sink<Object>> outputs;
  @SuppressWarnings(value = "VolatileArrayField")
  protected volatile Sink<Object>[] sinks = Sink.NO_SINKS;
  protected boolean alive;
  protected final OPERATOR operator;
  protected final PortMappingDescriptor descriptor;
  public long currentWindowId;
  protected long endWindowEmitTime;
  protected long lastSampleCpuTime;
  protected ThreadMXBean tmb;
  protected HashMap<SweepableReservoir, Long> endWindowDequeueTimes; // end window dequeue time for input ports
  protected Checkpoint checkpoint;
  public int applicationWindowCount;
  public int checkpointWindowCount;
  public int nextCheckpointWindowCount;
  public int dagCheckpointOffsetCount;
  protected int controlTupleCount;
  public final OperatorContext context;
  public final BlockingQueue<StatsListener.OperatorResponse> commandResponse;
  private final List<Field> metricFields;
  private final Map<String, Method> metricMethods;
  private ExecutorService executorService;
  private Queue<Pair<FutureTask<Stats.CheckpointStats>, CheckpointWindowInfo>> taskQueue;
  protected Stats.CheckpointStats checkpointStats;
  public long firstWindowMillis;
  public long windowWidthMillis;

  protected boolean reuseOperator;

  public Node(OPERATOR operator, OperatorContext context)
  {
    this.operator = operator;
    this.context = context;
    executorService = Executors.newSingleThreadExecutor();
    taskQueue = new LinkedList<>();

    outputs = new HashMap<>();

    descriptor = new PortMappingDescriptor();
    Operators.describe(operator, descriptor);

    endWindowDequeueTimes = new HashMap<>();
    tmb = ManagementFactory.getThreadMXBean();
    commandResponse = new LinkedBlockingQueue<>();

    metricFields = Lists.newArrayList();
    for (Field field : ReflectionUtils.getDeclaredFieldsIncludingInherited(operator.getClass())) {
      if (field.isAnnotationPresent(AutoMetric.class)) {
        metricFields.add(field);
        field.setAccessible(true);
      }
    }

    metricMethods = Maps.newHashMap();
    try {
      for (PropertyDescriptor pd : Introspector.getBeanInfo(operator.getClass()).getPropertyDescriptors()) {
        Method readMethod = pd.getReadMethod();
        if (readMethod != null) {
          AutoMetric rfa = readMethod.getAnnotation(AutoMetric.class);
          if (rfa != null) {
            metricMethods.put(pd.getName(), readMethod);
          }
        }
      }
    } catch (IntrospectionException e) {
      throw new RuntimeException("introspecting {}", e);
    }
  }

  public Operator getOperator()
  {
    return operator;
  }

  public void setReuseOperator(boolean reuseOperator)
  {
    this.reuseOperator = reuseOperator;
  }

  @Override
  public void setup(OperatorContext context)
  {
    shutdown = false;
    if (!reuseOperator) {
      logger.debug("Operator Context = {}", context);
      operator.setup(context);
    }
//    this is where the ports should be setup but since the
//    portcontext is not available here, we are doing it in
//    StramChild. In future version, we should move that code here
//    for (InputPort<?> port : descriptor.inputPorts.values()) {
//      port.setup(null);
//    }
//
//    for (OutputPort<?> port : descriptor.outputPorts.values()) {
//      port.setup(null);
//    }
  }

  @Override
  public void teardown()
  {
    for (PortContextPair<InputPort<?>> pcpair : descriptor.inputPorts.values()) {
      pcpair.component.teardown();
    }

    for (PortContextPair<OutputPort<?>> pcpair : descriptor.outputPorts.values()) {
      pcpair.component.teardown();
    }

    if (executorService != null) {
      executorService.shutdown();
      boolean terminated = false;
      try {
        terminated = executorService.awaitTermination(100, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        logger.debug("Wait for graceful executor service {} shutdown interrupted for node {}", executorService, this, e);
      }
      if (!terminated) {
        logger.warn("Shutting down executor service {} for node {}", executorService, this);
        executorService.shutdownNow();
      }
    }
    operator.teardown();
  }

  public PortMappingDescriptor getPortMappingDescriptor()
  {
    return descriptor;
  }

  public void connectOutputPort(String port, final Sink<Object> sink)
  {
    PortContextPair<OutputPort<?>> outputPort = descriptor.outputPorts.get(port);
    if (outputPort != null) {
      if (sink == null) {
        outputPort.component.setSink(null);
        outputs.remove(port);
      } else {
        outputPort.component.setSink(sink);
        outputs.put(port, sink);
      }
    }
  }

  public abstract void connectInputPort(String port, final SweepableReservoir reservoir);

  @SuppressWarnings({"unchecked"})
  public void addSinks(Map<String, Sink<Object>> sinks)
  {
    boolean changes = false;
    for (Entry<String, Sink<Object>> e : sinks.entrySet()) {
      /* make sure that we ignore all the input ports */
      PortContextPair<OutputPort<?>> pcpair = descriptor.outputPorts.get(e.getKey());
      if (pcpair == null) {
        continue;
      }
      changes = true;

      Sink<Object> ics = outputs.get(e.getKey());
      if (ics == null) {
        pcpair.component.setSink(e.getValue());
        outputs.put(e.getKey(), e.getValue());
        changes = true;
      } else if (ics instanceof MuxSink) {
        ((MuxSink)ics).add(e.getValue());
      } else {
        MuxSink muxSink = new MuxSink(ics, e.getValue());
        pcpair.component.setSink(muxSink);
        outputs.put(e.getKey(), muxSink);
        changes = true;
      }
    }

    if (changes) {
      activateSinks();
    }
  }

  public void removeSinks(Map<String, Sink<Object>> sinks)
  {
    boolean changes = false;
    for (Entry<String, Sink<Object>> e : sinks.entrySet()) {
      /* make sure that we ignore all the input ports */
      PortContextPair<OutputPort<?>> pcpair = descriptor.outputPorts.get(e.getKey());
      if (pcpair == null) {
        continue;
      }

      Sink<Object> ics = outputs.get(e.getKey());
      if (ics == e.getValue()) {
        pcpair.component.setSink(null);
        outputs.remove(e.getKey());
        changes = true;
      } else if (ics instanceof MuxSink) {
        MuxSink ms = (MuxSink)ics;
        ms.remove(e.getValue());
        Sink<Object>[] sinks1 = ms.getSinks();
        if (sinks1.length == 0) {
          pcpair.component.setSink(null);
          outputs.remove(e.getKey());
          changes = true;
        } else if (sinks1.length == 1) {
          pcpair.component.setSink(sinks1[0]);
          outputs.put(e.getKey(), sinks1[0]);
          changes = true;
        }
      }
    }

    if (changes) {
      activateSinks();
    }
  }

  protected ProcessingMode PROCESSING_MODE;
  protected volatile boolean shutdown;

  /**
   * Shutdown the current node.
   * If processEndWindow is set to true, then after shutting down, endWindow is called and
   * END_STREAM tuple is sent to downstream operators.
   *
   * @param processEndWindow
   */
  public void shutdown(boolean processEndWindow)
  {
    if (!processEndWindow) {
      shutdown = true;
    }

    synchronized (this) {
      alive = false;
    }

    if (context == null) {
      logger.warn("Shutdown requested when context is not available!");
    } else {
      /*
       * Since alive is non-volatile this code explicitly unsets it in the operator lifecycle theread thereby notifying
       * it even when the thread is reading it from the cache
       */
      context.request(new OperatorRequest()
      {
        @Override
        public StatsListener.OperatorResponse execute(Operator operator, int operatorId, long windowId) throws IOException
        {
          alive = false;
          return null;
        }

      });
    }
  }

  public void shutdown()
  {
    shutdown(false);
  }

  @Override
  public String toString()
  {
    return String.valueOf(getId());
  }

  protected void emitEndStream()
  {
    logger.info("{} sending EndOfStream", this);
    /*
     * since we are going away, we should let all the downstream operators know that.
     */
    EndStreamTuple est = new EndStreamTuple(currentWindowId);
    for (final Sink<Object> output : outputs.values()) {
      output.put(est);
    }
    controlTupleCount++;
  }

  protected void emitEndWindow()
  {
    long windowId = (operator instanceof Operator.DelayOperator) ?
        WindowGenerator.getAheadWindowId(currentWindowId, firstWindowMillis, windowWidthMillis, 1) : currentWindowId;
    EndWindowTuple ewt = new EndWindowTuple(windowId);
    for (int s = sinks.length; s-- > 0; ) {
      sinks[s].put(ewt);
    }
    controlTupleCount++;
  }

  protected void handleRequests(long windowId)
  {
    /*
     * we prefer to cater to requests at the end of the window boundary.
     */
    try {
      BlockingQueue<OperatorRequest> requests = context.getRequests();
      int size;
      StatsListener.OperatorResponse response;
      if ((size = requests.size()) > 0) {
        while (size-- > 0) {
          //logger.debug("endwindow: " + t.getWindowId() + " lastprocessed: " + context.getLastProcessedWindowId());
          response = requests.remove().execute(operator, context.getId(), windowId);
          if (response != null) {
            commandResponse.add(response);
          }
        }
      }
    } catch (Error er) {
      throw er;
    } catch (RuntimeException re) {
      throw re;
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  protected Map<String, Object> collectMetrics()
  {
    if (context.areMetricsListed() && (context.metricsToSend == null || context.metricsToSend.isEmpty())) {
      return null;
    }
    Map<String, Object> metricValues = Maps.newHashMap();
    try {
      for (Field field : metricFields) {
        if (context.metricsToSend != null && !context.metricsToSend.contains(field.getName())) {
          continue;
        }
        metricValues.put(field.getName(), field.get(operator));
      }

      for (Map.Entry<String, Method> methodEntry : metricMethods.entrySet()) {
        if (context.metricsToSend != null && !context.metricsToSend.contains(methodEntry.getKey())) {
          continue;
        }
        metricValues.put(methodEntry.getKey(), methodEntry.getValue().invoke(operator));
      }

      context.clearMetrics();
      return metricValues;
    } catch (IllegalAccessException | InvocationTargetException iae) {
      throw new RuntimeException(iae);
    }
  }

  protected void reportStats(ContainerStats.OperatorStats stats, long windowId)
  {
    stats.outputPorts = new ArrayList<>();
    for (Entry<String, Sink<Object>> e : outputs.entrySet()) {
      ContainerStats.OperatorStats.PortStats portStats = new ContainerStats.OperatorStats.PortStats(e.getKey());
      portStats.tupleCount = e.getValue().getCount(true) - controlTupleCount;
      portStats.endWindowTimestamp = endWindowEmitTime;
      stats.outputPorts.add(portStats);
    }
    controlTupleCount = 0;

    long currentCpuTime = tmb.getCurrentThreadCpuTime();
    stats.cpuTimeUsed = currentCpuTime - lastSampleCpuTime;
    lastSampleCpuTime = currentCpuTime;

    if (checkpoint != null) {
      stats.checkpoint = checkpoint;
      stats.checkpointStats = checkpointStats;
      checkpointStats = null;
      checkpoint = null;
    } else {
      Pair<FutureTask<Stats.CheckpointStats>, CheckpointWindowInfo> pair = taskQueue.peek();
      if (pair != null && pair.getFirst().isDone()) {
        taskQueue.poll();
        try {
          CheckpointWindowInfo checkpointWindowInfo = pair.getSecond();
          stats.checkpointStats = pair.getFirst().get();
          stats.checkpoint = new Checkpoint(checkpointWindowInfo.windowId, checkpointWindowInfo.applicationWindowCount,
              checkpointWindowInfo.checkpointWindowCount);
          if (operator instanceof Operator.CheckpointListener) {
            ((Operator.CheckpointListener)operator).checkpointed(checkpointWindowInfo.windowId);
          }
        } catch (Exception ex) {
          throw Throwables.propagate(ex);
        }
      }
    }

    context.report(stats, windowId);
  }

  protected void activateSinks()
  {
    int size = outputs.size();
    if (size == 0) {
      sinks = Sink.NO_SINKS;
    } else {
      @SuppressWarnings("unchecked")
      Sink<Object>[] newSinks = (Sink<Object>[])Array.newInstance(Sink.class, size);
      for (Sink<Object> s : outputs.values()) {
        newSinks[--size] = s;
      }

      sinks = newSinks;
    }
  }

  protected void deactivateSinks()
  {
    sinks = Sink.NO_SINKS;
  }

  void checkpoint(long windowId)
  {
    if (!context.stateless) {
      if (operator instanceof Operator.CheckpointNotificationListener) {
        ((Operator.CheckpointNotificationListener)operator).beforeCheckpoint(windowId);
      }

      StorageAgent ba = context.getValue(OperatorContext.STORAGE_AGENT);
      if (ba != null) {
        try {
          checkpointStats = new Stats.CheckpointStats();
          checkpointStats.checkpointStartTime = System.currentTimeMillis();
          ba.save(operator, id, windowId);
          if (ba instanceof AsyncStorageAgent) {
            AsyncStorageAgent asyncStorageAgent = (AsyncStorageAgent)ba;
            if (!asyncStorageAgent.isSyncCheckpoint()) {
              if (PROCESSING_MODE != ProcessingMode.EXACTLY_ONCE) {
                CheckpointWindowInfo checkpointWindowInfo = new CheckpointWindowInfo();
                checkpointWindowInfo.windowId = windowId;
                checkpointWindowInfo.applicationWindowCount = applicationWindowCount;
                checkpointWindowInfo.checkpointWindowCount = checkpointWindowCount;
                CheckpointHandler checkpointHandler = new CheckpointHandler();
                checkpointHandler.agent = asyncStorageAgent;
                checkpointHandler.operatorId = id;
                checkpointHandler.windowId = windowId;
                checkpointHandler.stats = checkpointStats;
                FutureTask<Stats.CheckpointStats> futureTask = new FutureTask<>(checkpointHandler);
                taskQueue.add(new Pair<>(futureTask, checkpointWindowInfo));
                executorService.submit(futureTask);
                checkpoint = null;
                checkpointStats = null;
                return;
              } else {
                asyncStorageAgent.flush(id, windowId);
              }
            }
          }
          checkpointStats.checkpointTime = System.currentTimeMillis() - checkpointStats.checkpointStartTime;
        } catch (IOException ie) {
          try {
            logger.warn("Rolling back checkpoint {} for Operator {} due to the exception {}",
                Codec.getStringWindowId(windowId), operator, ie);
            ba.delete(id, windowId);
          } catch (IOException ex) {
            logger.warn("Error while rolling back checkpoint", ex);
          }
          throw new RuntimeException(ie);
        }
      }
    }

    calculateNextCheckpointWindow();
    dagCheckpointOffsetCount = 0;
    checkpoint = new Checkpoint(windowId, applicationWindowCount, checkpointWindowCount);
    if (operator instanceof Operator.CheckpointListener) {
      ((Operator.CheckpointListener)operator).checkpointed(windowId);
    }
  }

  protected void calculateNextCheckpointWindow()
  {
    if (PROCESSING_MODE != ProcessingMode.EXACTLY_ONCE) {
      nextCheckpointWindowCount = ((DAG_CHECKPOINT_WINDOW_COUNT - dagCheckpointOffsetCount + CHECKPOINT_WINDOW_COUNT - 1) / CHECKPOINT_WINDOW_COUNT) * CHECKPOINT_WINDOW_COUNT;
    } else {
      nextCheckpointWindowCount = 1;
    }
  }

  @SuppressWarnings("unchecked")
  public static Node<?> retrieveNode(Object operator, OperatorContext context, OperatorDeployInfo.OperatorType type)
  {
    logger.debug("type={}, operator class={}", type, operator.getClass());

    Node<?> node;
    if (operator instanceof InputOperator && type == OperatorDeployInfo.OperatorType.INPUT) {
      node = new InputNode((InputOperator)operator, context);
    } else if (operator instanceof Unifier && type == OperatorDeployInfo.OperatorType.UNIFIER) {
      node = new UnifierNode((Unifier<Object>)operator, context);
    } else if (type == OperatorDeployInfo.OperatorType.OIO) {
      node = new OiONode((Operator)operator, context);
    } else {
      node = new GenericNode((Operator)operator, context);
    }

    return node;
  }

  /**
   * @return the id
   */
  public int getId()
  {
    return id;
  }

  /**
   * @param id the id to set
   */
  public void setId(int id)
  {
    if (this.id == 0) {
      this.id = id;
    } else {
      throw new RuntimeException("Id cannot be changed from " + this.id + " to " + id);
    }
  }

  @SuppressWarnings("unchecked")
  public void activate()
  {
    alive = true;
    APPLICATION_WINDOW_COUNT = context.getValue(OperatorContext.APPLICATION_WINDOW_COUNT);
    if (context.getValue(OperatorContext.SLIDE_BY_WINDOW_COUNT) != null) {
      int slidingWindowCount = context.getValue(OperatorContext.SLIDE_BY_WINDOW_COUNT);
      APPLICATION_WINDOW_COUNT = IntMath.gcd(APPLICATION_WINDOW_COUNT, slidingWindowCount);
    }
    DAG_CHECKPOINT_WINDOW_COUNT = context.getValue(Context.DAGContext.CHECKPOINT_WINDOW_COUNT);
    CHECKPOINT_WINDOW_COUNT = context.getValue(OperatorContext.CHECKPOINT_WINDOW_COUNT);
    Collection<StatsListener> statsListeners = context.getValue(OperatorContext.STATS_LISTENERS);

    if (CHECKPOINT_WINDOW_COUNT % APPLICATION_WINDOW_COUNT != 0) {
      logger.warn("{} is not exact multiple of {} for operator {}. This may cause side effects such as processing to begin without beginWindow preceding it in the first window after activation.",
          OperatorContext.CHECKPOINT_WINDOW_COUNT,
          OperatorContext.APPLICATION_WINDOW_COUNT,
          operator);
    }

    PROCESSING_MODE = context.getValue(OperatorContext.PROCESSING_MODE);
    if (PROCESSING_MODE == ProcessingMode.EXACTLY_ONCE && CHECKPOINT_WINDOW_COUNT != 1) {
      logger.warn("Ignoring {} attribute in favor of {} processing mode", OperatorContext.CHECKPOINT_WINDOW_COUNT.getSimpleName(), ProcessingMode.EXACTLY_ONCE.name());
      CHECKPOINT_WINDOW_COUNT = 1;
    }

    activateSinks();
    if (operator instanceof Operator.ActivationListener) {
      ((Operator.ActivationListener<OperatorContext>)operator).activate(context);
    }

    if (statsListeners != null) {
      Iterator<StatsListener> iterator = statsListeners.iterator();
      while (iterator.hasNext()) {
        DATA_TUPLE_AWARE = iterator.next().getClass().isAnnotationPresent(StatsListener.DataQueueSize.class);
        if (DATA_TUPLE_AWARE) {
          break;
        }
      }
    }

    if (!DATA_TUPLE_AWARE && (operator instanceof StatsListener)) {
      DATA_TUPLE_AWARE = operator.getClass().isAnnotationPresent(StatsListener.DataQueueSize.class);
    }
    /*
     * If there were any requests which needed to be executed before the operator started
     * its normal execution, execute those requests now - e.g. Restarting the operator
     * recording for the operators which failed while recording and being replaced.
     */
    handleRequests(currentWindowId);
  }

  public void deactivate()
  {
    if (operator instanceof Operator.ActivationListener) {
      ((Operator.ActivationListener<?>)operator).deactivate();
    }

    if (!shutdown && !alive) {
      emitEndStream();
    }

    deactivateSinks();
  }

  private class CheckpointHandler implements Callable<Stats.CheckpointStats>
  {
    public AsyncStorageAgent agent;
    public int operatorId;
    public long windowId;
    public Stats.CheckpointStats stats;

    @Override
    public Stats.CheckpointStats call() throws Exception
    {
      agent.flush(id, windowId);
      stats.checkpointTime = System.currentTimeMillis() - stats.checkpointStartTime;
      return stats;
    }
  }

  private class CheckpointWindowInfo
  {
    public int applicationWindowCount;
    public int checkpointWindowCount;
    public long windowId;
  }

  private static final Logger logger = LoggerFactory.getLogger(Node.class);
}