/*

  * 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.job.JobProgressTracker;

 import org.apache.giraph.utils.ThreadUtils;

 import org.apache.log4j.Logger;

 /**

  * Class which periodically writes worker's progress to zookeeper

  */

 public class WorkerProgressWriter {

   /** Class logger */

   private static final Logger LOG =

       Logger.getLogger(WorkerProgressWriter.class);

   /** How often to update worker's progress */

   private static final int WRITE_UPDATE_PERIOD_MILLISECONDS = 10 * 1000;

   /** Job progress tracker */

   private final JobProgressTracker jobProgressTracker;

   /** Thread which writes worker's progress */

   private final Thread writerThread;

   /** Whether worker finished application */

   private volatile boolean finished = false;

   /**

    * Constructor, starts separate thread to periodically update worker's

    * progress

    *

    * @param jobProgressTracker JobProgressTracker to report job progress to

    */

   public WorkerProgressWriter(JobProgressTracker jobProgressTracker) {

     this.jobProgressTracker = jobProgressTracker;

     writerThread = ThreadUtils.startThread(new Runnable() {

       @Override

       public void run() {

         while (!finished) {

           updateAndSendProgress();

           double factor = 1 + Math.random();

           if (!ThreadUtils.trySleep(

               (long) (WRITE_UPDATE_PERIOD_MILLISECONDS * factor))) {

             break;

           }

         }

       }

     }, "workerProgressThread");

   }

   /**

    * Update worker progress and send it

    */

   private void updateAndSendProgress() {

     WorkerProgress.get().updateMemory();

     jobProgressTracker.updateProgress(WorkerProgress.get());

   }

   /**

    * Stop the thread which writes worker's progress

    */

   public void stop() throws InterruptedException {

     finished = true;

     writerThread.interrupt();

     writerThread.join();

     updateAndSendProgress();

   }

 }