/*

  * 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.worker;

 import org.apache.giraph.conf.BooleanConfOption;

 import org.apache.giraph.conf.FloatConfOption;

 import org.apache.giraph.conf.GiraphConfiguration;

 import org.apache.giraph.conf.IntConfOption;

 import org.apache.giraph.utils.MemoryUtils;

 import org.apache.giraph.utils.ThreadUtils;

 import org.apache.giraph.zk.ZooKeeperExt;

 import org.apache.log4j.Logger;

 import org.apache.zookeeper.CreateMode;

 import org.apache.zookeeper.KeeperException;

 import org.apache.zookeeper.ZooDefs;

 import java.util.concurrent.atomic.AtomicLong;

 /**

  * Memory observer to help synchronize when full gcs are happening across all

  * the workers

  */

 public class MemoryObserver {

   /** Whether or not to use memory observer */

   public static final BooleanConfOption USE_MEMORY_OBSERVER =

       new BooleanConfOption("giraph.memoryObserver.enabled", false,

           "Whether or not to use memory observer");

   /** For which fraction of free memory will we issue manual gc calls */

   public static final FloatConfOption FREE_MEMORY_FRACTION_FOR_GC =

       new FloatConfOption("giraph.memoryObserver.freeMemoryFractionForGc", 0.1f,

           "For which fraction of free memory will we issue manual gc calls");

   /** Minimum milliseconds between two manual gc calls */

   public static final IntConfOption MIN_MS_BETWEEN_FULL_GCS =

       new IntConfOption("giraph.memoryObserver.minMsBetweenFullGcs", 60 * 1000,

           "Minimum milliseconds between two manual gc calls");

   /** Logger */

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

   /** How long does memory observer thread sleep for */

   private static final int MEMORY_OBSERVER_SLEEP_MS = 1000;

   /** When was the last manual gc call */

   private final AtomicLong lastManualGc = new AtomicLong();

   /** Zookeeper */

   private final ZooKeeperExt zk;

   /** Path on zookeeper for memory observer files */

   private final String zkPath;

   /** Value of conf setting MIN_MS_BETWEEN_FULL_GCS */

   private final int minMsBetweenFullGcs;

   /**

    * Constructor

    *

    * @param zk Zookeeper

    * @param zkPath Path on zookeeper for memory observer files

    * @param conf Configration

    */

   public MemoryObserver(final ZooKeeperExt zk,

       final String zkPath, GiraphConfiguration conf) {

     this.zk = zk;

     this.zkPath = zkPath;

     minMsBetweenFullGcs = MIN_MS_BETWEEN_FULL_GCS.get(conf);

     if (!USE_MEMORY_OBSERVER.get(conf)) {

       return;

     }

     try {

       // Create base path for memory observer nodes

       zk.createOnceExt(zkPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,

           CreateMode.PERSISTENT, true);

     } catch (KeeperException | InterruptedException e) {

       LOG.info("Exception occurred", e);

     }

     setWatcher();

     final float freeMemoryFractionForGc =

         FREE_MEMORY_FRACTION_FOR_GC.get(conf);

     ThreadUtils.startThread(new Runnable() {

       @Override

       public void run() {

         while (true) {

           double freeMemoryFraction = MemoryUtils.freeMemoryFraction();

           long msFromLastGc = System.currentTimeMillis() - lastManualGc.get();

           if (msFromLastGc > minMsBetweenFullGcs &&

               freeMemoryFraction < freeMemoryFractionForGc) {

             try {

               if (LOG.isInfoEnabled()) {

                 LOG.info("Notifying others about low memory (" +

                     freeMemoryFraction + "% free)");

               }

               zk.createExt(

                   zkPath + "/" + System.currentTimeMillis(),

                   new byte[0],

                   ZooDefs.Ids.OPEN_ACL_UNSAFE,

                   CreateMode.EPHEMERAL,

                   false);

             } catch (KeeperException | InterruptedException e) {

               LOG.warn("Exception occurred", e);

             }

           }

           if (!ThreadUtils.trySleep(MEMORY_OBSERVER_SLEEP_MS)) {

             return;

           }

         }

       }

     }, "memory-observer");

   }

   /** Set watcher on memory observer folder */

   private void setWatcher() {

     try {

       // Set a watcher on this path

       zk.getChildrenExt(zkPath, true, false, false);

     } catch (KeeperException | InterruptedException e) {

       LOG.warn("Exception occurred", e);

     }

   }

   /** Manually call gc, if enough time from last call has passed */

   public void callGc() {

     long last = lastManualGc.get();

     if (System.currentTimeMillis() - last > minMsBetweenFullGcs &&

         lastManualGc.compareAndSet(last, System.currentTimeMillis())) {

       if (LOG.isInfoEnabled()) {

         LOG.info("Calling gc manually");

       }

       System.gc();

       if (LOG.isInfoEnabled()) {

         LOG.info("Manual gc call done");

       }

     }

     setWatcher();

   }

 }