package com.conveyal.r5.analyst.broker;

import gnu.trove.iterator.TIntLongIterator;
import gnu.trove.map.TIntLongMap;
import gnu.trove.map.TIntObjectMap;
import gnu.trove.map.hash.TIntLongHashMap;
import gnu.trove.map.hash.TIntObjectHashMap;
import gnu.trove.set.TIntSet;
import gnu.trove.set.hash.TIntHashSet;
import com.conveyal.r5.analyst.cluster.GenericClusterRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayDeque;
import java.util.List;
import java.util.Queue;

/**
 * FIXME delivered tasks map is oblivious to multiple tasks having the same ID.
 * In fact we just generate numeric queue task IDs. Origin point IDs will be handled at the application layer.
 * FIXME ^^ Does this mean we "should" generate numeric task IDs?
 */
public class Job {

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

    public static final int REDELIVERY_QUIET_PERIOD_MSEC = 60 * 1000;

    /* A unique identifier for this job, usually a random UUID. */
    public final String jobId;

    /* The graph on which tasks are to be run. All tasks contained in a job must run on the same graph. */
    String graphId;

    /* Tasks in this job that have yet to be delivered, or that will be re-delivered due to completion timeout. */
    // maybe this should only be a list of IDs.
    Queue<GenericClusterRequest> tasksAwaitingDelivery = new ArrayDeque<>();

    /* The tasks in this job keyed on their task ID. */
    TIntObjectMap<GenericClusterRequest> tasksById = new TIntObjectHashMap<>();

    /* The last time tasks were delivered to a worker. Enables a quiet period between delivery and re-delivery. */
    long lastDeliveryTime = 0;

    /* The IDs of all tasks that have been marked completed. */
    TIntSet completedTasks = new TIntHashSet();

    /* How many times incomplete tasks were redelivered on this job. */
    public int redeliveryCount = 0;

    public Job (String jobId) {
        this.jobId = jobId;
    }

    /** Adds a task to this Job, assigning it a task ID number. */
    public void addTask (GenericClusterRequest task) {
        tasksById.put(task.taskId, task);
        tasksAwaitingDelivery.add(task);
    }

    /**
     * Check if there are any delivered tasks that were never marked as completed.
     * This could happen if the workers are spot instances and they are terminated by AWS while processing some tasks.
     */
    public int redeliver () {

        // Only redeliver on jobs that are still active (that have some tasks not marked complete).
        if (this.isComplete()) return 0;

        // Only redeliver tasks when the delivery queue is empty.
        // This reduces the frequency of redelivery operations and is more predictable.
        if (tasksAwaitingDelivery.size() > 0) return 0;

        // After the last task is delivered, wait a while before redelivering to avoid spurious re-delivery.
        if (System.currentTimeMillis() - lastDeliveryTime < REDELIVERY_QUIET_PERIOD_MSEC) return 0;

        // If we arrive here, we should have an empty delivery queue but some tasks that are not marked complete.
        tasksById.forEachEntry((taskId, task) -> {
            if (!completedTasks.contains(taskId)) {
                tasksAwaitingDelivery.add(task);
            }
            return true;
        });
        redeliveryCount += 1;
        LOG.info("Redelivered {} incomplete tasks on job {}.", tasksAwaitingDelivery.size(), this.jobId);
        return tasksAwaitingDelivery.size();
    }

    public boolean isComplete() {
        return completedTasks.size() == tasksById.size();
    }

    public boolean containsTask (int taskId) {
        GenericClusterRequest req = tasksById.get(taskId);
        if (req != null) {
            if (!req.jobId.equals(this.jobId)) {
                LOG.error("Task {} has a job ID that does not match the job in which it was discovered.");
            }
            return true;
        }
        return false;
    }

}

@@ -929,24 +929,75 @@
 60 * 1000;%0A%0A
+    public static final int MAX_REDELIVERIES = 2;%0A%0A
     /* A uni
@@ -3416,23 +3416,23 @@
 info(%22Re
-deliver
+-enqueu
 ed %7B%7D in
@@ -3446,16 +3446,29 @@
 e tasks 
+for delivery 
 on job %7B
@@ -3552,32 +3552,33 @@
 elivery.size();%0A
+%0A
     %7D%0A%0A    publi

change log text