Import chunks on the main thread.

ImportChunksTask.java

package com.comphenix.example;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.bukkit.Location;
import org.bukkit.plugin.Plugin;
import org.bukkit.scheduler.BukkitScheduler;

/**
 * Represents a task for importing chunk data on the main thread.
 * 
 * @author Kristian
 */
public class ImportChunksTask {
    /**
     * The chunk we are importing. 
     * 
     * @author Kristian
     */
    public static class PendingChunk {
        // Origin block
        private final Location origin;
        
        private final int widthX;
        private final int widthZ;
                
        private final byte[] blocks;

        /**
         * Construct a new pending chunk operation. A chunk is a NxMxH cube of blocks.
         * <p>
         * Note that the height is implicitly deduced from the full length of the block array.
         * <p>
         * The origin block is defined as the first block in the block array
         * @param origin - absolute position of the origin block.
         * @param widthX - the width (N) in the x-axis.
         * @param widthZ - the width (M) in the z-axis.
         * @param blocks - array of block IDs.
         */
        public PendingChunk(Location origin, int widthX, int widthZ, byte[] blocks) {
            this.origin = origin;
            this.widthX = widthX;
            this.widthZ = widthZ;
            this.blocks = blocks;
        }

        public Location getOrigin() {
            return origin;
        }

        public int getWidthX() {
            return widthX;
        }

        public int getWidthZ() {
            return widthZ;
        }

        public byte[] getBlocks() {
            return blocks;
        }
    }
    
    private static class PendingOperation {
        private final Location lastLocation;
        private final int lastIndex;
        
        public PendingOperation(Location lastLocation, int lastIndex) {
            this.lastLocation = lastLocation;
            this.lastIndex = lastIndex;
        }

        public Location getLastLocation() {
            return lastLocation;
        }

        public int getLastIndex() {
            return lastIndex;
        }
    }
    
    /**
     * This is a queue of chunks that are waiting to be processed. 
     * <p>
     * It is unbounded, but you could use a different concurrent queue that has a limit or
     * better performance characteristics.
     */
    private final Queue<PendingChunk> pendingChunks = new ConcurrentLinkedQueue<PendingChunk>();
    
    /**
     * This is the chunk we are currently processing.
     */
    private PendingChunk lastChunk;
    
    /**
     * Saved state from the last tick.
     */
    private PendingOperation lastOperation;
    
    // Used to make mass block updates
    private MassBlockFactory updateFactory;
    
    // Maximum number of nanoseconds to consume per tick
    private long maximumTickConsumption;
    
    // Current task
    private int taskID = -1;
    
    public ImportChunksTask(MassBlockFactory updateFactory, long maximumTickConsumption) {
        this.updateFactory = updateFactory;
        this.maximumTickConsumption = maximumTickConsumption;
    }
    
    /**
     * Enqueue a chunk for import on the main thread.
     * @param chunk - the chunk to import.
     */
    public void queueImportChunk(PendingChunk chunk) {
        pendingChunks.add(chunk);
    }
    
    /**
     * Retrieve the maximum number of nanoseconds to consume per tick.
     * @return Maximum number of nanoseconds.
     */
    public long getMaximumTickConsumption() {
        return maximumTickConsumption;
    }
    
    /**
     * Start the import chunk task.
     * @param scheduler - the Bukkit scheduler.
     * @param plugin - the owner plugin.
     */
    public void start(BukkitScheduler scheduler, Plugin plugin) {
        if (taskID < 0) {
            throw new IllegalStateException("Task has already been started.");
        }
            
        // Start and save task ID
        taskID = scheduler.scheduleSyncDelayedTask(plugin, new Runnable() {
            @Override
            public void run() {
                processTick();
            }
        });
        
        if (taskID < 0) {
            throw new IllegalStateException("Unable to start task.");
        }
    }
    
    /**
     * Stop the task, if it is running.
     */
    public void stop(BukkitScheduler scheduler) {
        if (taskID >= 0) {
            scheduler.cancelTask(taskID);
            taskID = -1;
        }
    }

    private PendingChunk getPendingChunk() {
        if (lastChunk != null)
            return lastChunk;
        
        PendingChunk chunk = pendingChunks.poll();
        
        // Reset state
        if (chunk != null) {
            lastOperation = null;
        }
        return chunk;
    }
    
    /**
     * Invoked when we process the current tick.
     */
    private void processTick() {
        long startTime = System.nanoTime();
        
        // Process chunks until we're done
        while ((lastChunk = getPendingChunk()) != null) {
            lastOperation = processChunk(lastChunk, lastOperation, startTime);
            
            // If the processor saved its state, it's a cue for us to exit
            if (lastOperation != null) {
                return;
            }
        }
    }
    
    /**
     * Process a given chunk, using the previous state if provided.
     * @param chunk - the chunk to process/import.
     * @param operation - state from the previous operation, or NULL if this is the first tick.
     * @param startTime - time we started processing.
     * @return State to save until the next tick, or NULL if we completed it.
     */
    private PendingOperation processChunk(PendingChunk chunk, PendingOperation operation, long startTime) {
        byte[] blocks = chunk.getBlocks();
        
        // Absolute location of origin
        Location origin = chunk.getOrigin();
        int aX = origin.getBlockX();
        int aY = origin.getBlockY();
        int aZ = origin.getBlockZ();
        
        // Relative location from origin
        int dX = 0, dY = 0, dZ = 0;

        // Current index
        int index = operation != null ? operation.getLastIndex() : 0;
        
        // What will update all the blocks
        MassBlockUpdate updator = updateFactory.construct(origin.getWorld());
        
        // Load a previous location?
        if (operation != null) {
            dX = operation.getLastLocation().getBlockX() - aX;
            dY = operation.getLastLocation().getBlockY() - aY;
            dZ = operation.getLastLocation().getBlockZ() - aZ;
        }
        
        // Process as many blocks as possible
        for (; index < blocks.length; index++) {
            updator.setBlock(aX + dX, aY + dY, aZ + dZ, blocks[index]);
            
            // Increment location
            dX++;
            
            // Check bounds
            if (dX >= chunk.getWidthX()) {
                dX = 0; dZ++;
            }
            if (dZ >= chunk.getWidthZ()) {
                dZ = 0; dY++;
            }
            
            // Check timeout
            if (System.nanoTime() - startTime > maximumTickConsumption) {
                // Save this for the next tick
                operation = new PendingOperation(
                    new Location(origin.getWorld(), aX + dX, aY + dY, aZ + dZ),
                    index + 1);
                break;
            }
        }
        
        // We are done
        if (index >= blocks.length) {
            operation = null;
        }
        
        // Lighting calculation has to be done here, since other blocks may be modified in the future
        updator.notifyClients();
        return operation;
    }
}