code: package ca.uwaterloo.redynisdaemon.threads;
  
  import ca.uwaterloo.redynisdaemon.beans.UsageMetric;
  import ca.uwaterloo.redynisdaemon.exceptions.InternalAppError;
  import ca.uwaterloo.redynisdaemon.instructions.PlacementInstruction;
  import ca.uwaterloo.redynisdaemon.utils.Constants;
  import ca.uwaterloo.redynisdaemon.utils.Options;
  import ca.uwaterloo.redynisdaemon.utils.RedisHelper;
  import org.apache.logging.log4j.LogManager;
  import org.apache.logging.log4j.Logger;
  import org.joda.time.DateTime;
  
  import java.io.IOException;
  import java.util.*;
  import java.util.concurrent.ScheduledExecutorService;
  
  public class AnalyzerThread implements Runnable
  {
      private static Logger log = LogManager.getLogger(AnalyzerThread.class);
      private ScheduledExecutorService placementScheduler;
      private ScheduledExecutorService deletionScheduler;
  
      public AnalyzerThread(ScheduledExecutorService placementScheduler, ScheduledExecutorService deletionScheduler)
      {
          this.placementScheduler = placementScheduler;
          this.deletionScheduler = deletionScheduler;
      }
  
      @Override
      public void run()
      {
          try
          {
              log.info("Analyzer Thread begun");
              Set<PlacementInstruction> placementInstructions = new HashSet<>();
  
              Map<String, UsageMetric> allUsageMetrics = getAllUsageMetrics();
  
              PlacementInstruction instruction;
              for (Map.Entry<String, UsageMetric> entry : allUsageMetrics.entrySet())
              {
                  instruction = analyzeKeyMetrics(entry);
  
                  if (null != instruction)
                  {
                      placementInstructions.add(instruction);
                  }
              }
  
              if (placementInstructions.size() > 0)
              {
                  placementScheduler.execute(new PlacementThread(placementInstructions, deletionScheduler));
              }
          }
          catch (Exception e)
          {
              log.error("Encountered exception while executing Analyzer Thread. Terminating thread. ", e);
          }
  
          log.info("Analyzer Thread concluded");
      }
  
      private PlacementInstruction analyzeKeyMetrics(Map.Entry<String, UsageMetric> keyMetric)
          throws InternalAppError
      {
          PlacementInstruction instruction = null;
  
          String redisKey = keyMetric.getKey();
          UsageMetric usageMetric = keyMetric.getValue();
  
          DateTime lastAccessedDate = new DateTime(usageMetric.getLastAccessedDate().getTime());
          DateTime expiryDeadline =
              DateTime.now().minusSeconds(Options.getInstance().getAppConfig().getKeyExpirySeconds());
  
          // if last accessed date is way in the past, delete the key
          if (lastAccessedDate.isBefore(expiryDeadline))
          {
              log.debug("Redis key " + redisKey + " expired. Deleting from all owner hosts");
              instruction = new PlacementInstruction(redisKey, null, new HashSet<String>(), usageMetric.getHosts());
          }
          else
          {
              Map<String, Integer> hostAccesses = usageMetric.getHostAccesses();
              Set<String> oldHosts = usageMetric.getHosts();
              Set<String> currentHosts = new HashSet<>();
  
              String hostname;
              Integer hostAccessCount;
              for (Map.Entry<String, Integer> hostAccessEntry : hostAccesses.entrySet())
              {
                  hostname = hostAccessEntry.getKey();
                  hostAccessCount = hostAccessEntry.getValue();
  
                  log.debug("hostAccessEntry: " + hostAccessEntry);
                  if (exceedsAccessThreshold(hostAccessCount, usageMetric.getTotalAccessCount()))
                  {
                      currentHosts.add(hostname);
                  }
              }
  
              // Set difference gives use the new hosts to replicate on
              Set<String> newHosts = new HashSet<>(currentHosts);
              newHosts.removeAll(oldHosts);
  
              // Set difference gives use the hosts to delete from
              Set<String> deleteHosts = new HashSet<>(oldHosts);
              deleteHosts.removeAll(currentHosts);
  
              log.debug("oldHosts: " + oldHosts);
              log.debug("currentHosts: " + currentHosts);
              log.debug("newHosts: " + newHosts);
              log.debug("deleteHosts: " + deleteHosts);
  
              String sourceHost = oldHosts.iterator().next();
              if (newHosts.size() != 0 || deleteHosts.size() != 0)
              {
                  instruction = new PlacementInstruction(redisKey, sourceHost, newHosts, deleteHosts);
              }
          }
  
          return instruction;
      }
  
      private Boolean exceedsAccessThreshold(Integer accesses, Integer totalAccessCount)
          throws InternalAppError
      {
          Double factor = (double) accesses / (double) totalAccessCount;
          log.debug("factor: " + factor);
          return factor >= Options.getInstance().getAppConfig().getAccessThreshold();
      }
  
      private Map<String, UsageMetric> getAllUsageMetrics()
          throws InternalAppError, IOException
      {
          RedisHelper redisHelper = RedisHelper.getInstance();
          List<String> keys = new ArrayList<>(redisHelper.getAllKeys());
          Map<String, UsageMetric> usageMetricMap = new HashMap<>();
  
          List<String> usageMetrics = null;
          if (!keys.isEmpty())
          {
              usageMetrics = redisHelper.multiGet(keys);
              for (int i = 0; i < keys.size(); i++)
              {
                  usageMetricMap.put(keys.get(i), Constants.MAPPER.readValue(usageMetrics.get(i), UsageMetric.class));
              }
          }
  
          return usageMetricMap;
      }
  }
  