/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto.operator;

import com.facebook.presto.execution.TaskOutput;
import com.facebook.presto.operator.ExchangeOperator.ExchangeClientStatus;
import com.facebook.presto.util.CpuTimer;
import com.facebook.presto.util.CpuTimer.CpuDuration;
import com.google.common.base.Preconditions;
import io.airlift.units.DataSize;
import io.airlift.units.DataSize.Unit;
import io.airlift.units.Duration;

import javax.annotation.concurrent.NotThreadSafe;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadMXBean;
import java.util.List;

/**
 * This class is not thread safe, but the done state of the task is properly
 * propagated from the TaskOutput, which is thread safe, to this class.
 */
@NotThreadSafe
public class OperatorStats
{
    private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean();

    private final TaskOutput taskOutput;

    private long declaredSize;
    private long declaredPositions;

    private long completedDataSize;
    private long completedPositions;

    private CpuTimer cpuTimer = new CpuTimer();
    private long exchangeWaitTime;

    private boolean finished;

    public OperatorStats()
    {
        this.taskOutput = null;
    }

    public OperatorStats(TaskOutput taskOutput)
    {
        Preconditions.checkNotNull(taskOutput, "taskOutput is null");
        this.taskOutput = taskOutput;
    }

    public boolean isDone()
    {
        return finished || (taskOutput != null && taskOutput.getState().isDone());
    }

    public void addDeclaredSize(long bytes)
    {
        if (taskOutput == null) {
            return;
        }

        taskOutput.getStats().addInputDataSize(new DataSize(bytes, Unit.BYTE));
        declaredSize += bytes;
    }

    public void addCompletedDataSize(long bytes)
    {
        if (taskOutput == null) {
            return;
        }

        taskOutput.getStats().addCompletedDataSize(new DataSize(bytes, Unit.BYTE));
        completedDataSize += bytes;

        if (completedDataSize > declaredSize) {
            taskOutput.getStats().addInputDataSize(new DataSize(completedDataSize - declaredSize, Unit.BYTE));
            declaredSize = completedDataSize;
        }

        updateTimings();
    }

    public void addCompletedPositions(long positions)
    {
        if (taskOutput == null) {
            return;
        }

        taskOutput.getStats().addCompletedPositions(positions);
        completedPositions += positions;

        if (completedPositions > declaredPositions) {
            taskOutput.getStats().addInputPositions(completedPositions - declaredPositions);
            declaredPositions = completedPositions;
        }

        updateTimings();
    }

    public void addExchangeWaitTime(Duration duration)
    {
        if (taskOutput == null) {
            return;
        }
        taskOutput.getStats().addExchangeWaitTime(duration);
    }

    public void setExchangeStatus(List<ExchangeClientStatus> exchangeStatus)
    {
        if (taskOutput == null) {
            return;
        }
        taskOutput.getStats().setExchangeStatus(exchangeStatus);
    }

    public void start()
    {
        if (taskOutput == null) {
            return;
        }

        taskOutput.getStats().splitStarted();
        cpuTimer = new CpuTimer();
    }

    public void finish()
    {
        if (finished) {
            return;
        }
        finished = true;

        if (taskOutput == null) {
            return;
        }

        updateTimings();
        taskOutput.getStats().splitCompleted();
    }

    private void updateTimings()
    {
        if (taskOutput == null) {
            return;
        }

        CpuDuration splitTime = cpuTimer.startNewInterval();
        taskOutput.getStats().addSplitWallTime(splitTime.getWall());
        taskOutput.getStats().addSplitCpuTime(splitTime.getCpu());
        taskOutput.getStats().addSplitUserTime(splitTime.getUser());
    }
}

@@ -523,97 +523,8 @@
 fe;%0A
-import java.lang.management.ManagementFactory;%0Aimport java.lang.management.ThreadMXBean;%0A
 impo
@@ -542,16 +542,16 @@
 l.List;%0A
+
 %0A/**%0A * 
@@ -748,101 +748,8 @@
 s%0A%7B%0A
-    private static final ThreadMXBean THREAD_MX_BEAN = ManagementFactory.getThreadMXBean();%0A%0A
     
@@ -920,32 +920,32 @@
 etedPositions;%0A%0A
+
     private CpuT
@@ -979,43 +979,8 @@
 r();
-%0A    private long exchangeWaitTime;
 %0A%0A  

Remove unused fields