/*

  * 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.giraph.ooc;

 import com.yammer.metrics.core.Counter;

 import com.yammer.metrics.core.Histogram;

 import org.apache.giraph.metrics.GiraphMetrics;

 import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;

 import org.apache.giraph.metrics.SuperstepMetricsRegistry;

 import org.apache.giraph.ooc.command.IOCommand;

 import org.apache.giraph.ooc.command.LoadPartitionIOCommand;

 import org.apache.giraph.ooc.command.WaitIOCommand;

 import org.apache.log4j.Logger;

 import java.util.concurrent.Callable;

 /**

  * IO threads for out-of-core mechanism.

  */

 public class OutOfCoreIOCallable implements Callable<Void>,

     ResetSuperstepMetricsObserver {

   /** Name of Metric for number of bytes read from disk */

   public static final String BYTES_LOAD_FROM_DISK = "ooc-bytes-load";

   /** Name of Metric for number of bytes written to disk */

   public static final String BYTES_STORE_TO_DISK = "ooc-bytes-store";

   /** Name of Metric for size of loads */

   public static final String HISTOGRAM_LOAD_SIZE = "ooc-load-size-bytes";

   /** Name of Metric for size of stores */

   public static final String HISTOGRAM_STORE_SIZE = "ooc-store-size-bytes";

   /** Class logger. */

   private static final Logger LOG = Logger.getLogger(OutOfCoreIOCallable.class);

   /** Out-of-core engine */

   private final OutOfCoreEngine oocEngine;

   /** Thread id/Disk id */

   private final int diskId;

   /** How many bytes of data is read from disk */

   private Counter bytesReadPerSuperstep;

   /** How many bytes of data is written to disk */

   private Counter bytesWrittenPerSuperstep;

   /** Size of load IO commands */

   private Histogram histogramLoadSize;

   /** Size of store IO commands */

   private Histogram histogramStoreSize;

   /**

    * Constructor

    *

    * @param oocEngine out-of-core engine

    * @param diskId thread id/disk id

    */

   public OutOfCoreIOCallable(OutOfCoreEngine oocEngine, int diskId) {

     this.oocEngine = oocEngine;

     this.diskId = diskId;

     newSuperstep(GiraphMetrics.get().perSuperstep());

     GiraphMetrics.get().addSuperstepResetObserver(this);

   }

   @Override

   public Void call() throws Exception {

     while (true) {

       oocEngine.getSuperstepLock().readLock().lock();

       IOCommand command = oocEngine.getIOScheduler().getNextIOCommand(diskId);

       if (LOG.isDebugEnabled() && !(command instanceof WaitIOCommand)) {

         LOG.debug("call: thread " + diskId + "'s next IO command is: " +

             command);

       }

       if (command == null) {

         oocEngine.getSuperstepLock().readLock().unlock();

         break;

       }

       if (command instanceof WaitIOCommand) {

         oocEngine.getSuperstepLock().readLock().unlock();

       }

       boolean commandExecuted = false;

       long duration = 0;

       long bytes;

       // CHECKSTYLE: stop IllegalCatch

       try {

         long timeInGC = oocEngine.getServiceWorker().getGraphTaskManager()

             .getSuperstepGCTime();

         long startTime = System.currentTimeMillis();

         commandExecuted = command.execute();

         duration = System.currentTimeMillis() - startTime;

         timeInGC = oocEngine.getServiceWorker().getGraphTaskManager()

             .getSuperstepGCTime() - timeInGC;

         bytes = command.bytesTransferred();

         if (LOG.isDebugEnabled() && !(command instanceof WaitIOCommand)) {

           LOG.debug("call: thread " + diskId + "'s command " + command +

               " completed: bytes= " + bytes + ", duration=" + duration + ", " +

               "bandwidth=" + String.format("%.2f", (double) bytes / duration *

               1000 / 1024 / 1024) +

               ((command instanceof WaitIOCommand) ? "" :

                   (", bandwidth (excluding GC time)=" + String.format("%.2f",

                       (double) bytes / (duration - timeInGC) *

                           1000 / 1024 / 1024))));

         }

       } catch (Exception e) {

         throw new RuntimeException(

             "call: execution of IO command " + command + " failed!", e);

       }

       // CHECKSTYLE: resume IllegalCatch

       if (!(command instanceof WaitIOCommand)) {

         oocEngine.getSuperstepLock().readLock().unlock();

         if (bytes != 0) {

           if (command instanceof LoadPartitionIOCommand) {

             bytesReadPerSuperstep.inc(bytes);

             histogramLoadSize.update(bytes);

           } else {

             bytesWrittenPerSuperstep.inc(bytes);

             histogramStoreSize.update(bytes);

           }

         }

       }

       if (commandExecuted && duration > 0) {

         oocEngine.getIOStatistics().update(command.getType(),

             command.bytesTransferred(), duration);

       }

       oocEngine.getIOScheduler().ioCommandCompleted(command);

     }

     if (LOG.isInfoEnabled()) {

       LOG.info("call: out-of-core IO thread " + diskId + " terminating!");

     }

     return null;

   }

   @Override

   public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {

     bytesReadPerSuperstep = superstepMetrics.getCounter(BYTES_LOAD_FROM_DISK);

     bytesWrittenPerSuperstep =

         superstepMetrics.getCounter(BYTES_STORE_TO_DISK);

     histogramLoadSize =

         superstepMetrics.getUniformHistogram(HISTOGRAM_LOAD_SIZE);

     histogramStoreSize =

         superstepMetrics.getUniformHistogram(HISTOGRAM_STORE_SIZE);

   }

 }