001/*
002 * PlotSquared, a land and world management plugin for Minecraft.
003 * Copyright (C) IntellectualSites <https://intellectualsites.com>
004 * Copyright (C) IntellectualSites team and contributors
005 *
006 * This program is free software: you can redistribute it and/or modify
007 * it under the terms of the GNU General Public License as published by
008 * the Free Software Foundation, either version 3 of the License, or
009 * (at your option) any later version.
010 *
011 * This program is distributed in the hope that it will be useful,
012 * but WITHOUT ANY WARRANTY; without even the implied warranty of
013 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
014 * GNU General Public License for more details.
015 *
016 * You should have received a copy of the GNU General Public License
017 * along with this program.  If not, see <https://www.gnu.org/licenses/>.
018 */
019package com.plotsquared.bukkit.queue;
020
021import com.google.inject.Inject;
022import com.google.inject.assistedinject.Assisted;
023import com.plotsquared.bukkit.BukkitPlatform;
024import com.plotsquared.core.PlotSquared;
025import com.plotsquared.core.queue.ChunkCoordinator;
026import com.plotsquared.core.queue.subscriber.ProgressSubscriber;
027import com.plotsquared.core.util.task.PlotSquaredTask;
028import com.plotsquared.core.util.task.TaskManager;
029import com.plotsquared.core.util.task.TaskTime;
030import com.sk89q.worldedit.math.BlockVector2;
031import com.sk89q.worldedit.world.World;
032import io.papermc.lib.PaperLib;
033import org.apache.logging.log4j.LogManager;
034import org.apache.logging.log4j.Logger;
035import org.bukkit.Bukkit;
036import org.bukkit.Chunk;
037import org.bukkit.plugin.Plugin;
038import org.bukkit.plugin.java.JavaPlugin;
039import org.checkerframework.checker.nullness.qual.NonNull;
040
041import java.util.Collection;
042import java.util.LinkedList;
043import java.util.List;
044import java.util.Queue;
045import java.util.concurrent.LinkedBlockingQueue;
046import java.util.concurrent.TimeUnit;
047import java.util.concurrent.atomic.AtomicInteger;
048import java.util.function.Consumer;
049
050/**
051 * Utility that allows for the loading and coordination of chunk actions
052 * <p>
053 * The coordinator takes in collection of chunk coordinates, loads them
054 * and allows the caller to specify a sink for the loaded chunks. The
055 * coordinator will prevent the chunks from being unloaded until the sink
056 * has fully consumed the chunk
057 * </p>
058 **/
059public final class BukkitChunkCoordinator extends ChunkCoordinator {
060
061    private static final Logger LOGGER = LogManager.getLogger("PlotSquared/" + BukkitChunkCoordinator.class.getSimpleName());
062
063    private final List<ProgressSubscriber> progressSubscribers = new LinkedList<>();
064
065    private final Queue<BlockVector2> requestedChunks;
066    private final Queue<Chunk> availableChunks;
067    private final long maxIterationTime;
068    private final Plugin plugin;
069    private final Consumer<BlockVector2> chunkConsumer;
070    private final org.bukkit.World bukkitWorld;
071    private final Runnable whenDone;
072    private final Consumer<Throwable> throwableConsumer;
073    private final boolean unloadAfter;
074    private final int totalSize;
075    private final AtomicInteger expectedSize;
076    private final AtomicInteger loadingChunks = new AtomicInteger();
077    private final boolean forceSync;
078    private final boolean shouldGen;
079
080    private int batchSize;
081    private PlotSquaredTask task;
082    private volatile boolean shouldCancel;
083    private boolean finished;
084
085    @Inject
086    private BukkitChunkCoordinator(
087            @Assisted final long maxIterationTime,
088            @Assisted final int initialBatchSize,
089            @Assisted final @NonNull Consumer<BlockVector2> chunkConsumer,
090            @Assisted final @NonNull World world,
091            @Assisted final @NonNull Collection<BlockVector2> requestedChunks,
092            @Assisted final @NonNull Runnable whenDone,
093            @Assisted final @NonNull Consumer<Throwable> throwableConsumer,
094            @Assisted("unloadAfter") final boolean unloadAfter,
095            @Assisted final @NonNull Collection<ProgressSubscriber> progressSubscribers,
096            @Assisted("forceSync") final boolean forceSync,
097            @Assisted("shouldGen") final boolean shouldGen
098    ) {
099        this.requestedChunks = new LinkedBlockingQueue<>(requestedChunks);
100        this.availableChunks = new LinkedBlockingQueue<>();
101        this.totalSize = requestedChunks.size();
102        this.expectedSize = new AtomicInteger(this.totalSize);
103        this.batchSize = initialBatchSize;
104        this.chunkConsumer = chunkConsumer;
105        this.maxIterationTime = maxIterationTime;
106        this.whenDone = whenDone;
107        this.throwableConsumer = throwableConsumer;
108        this.unloadAfter = unloadAfter;
109        this.plugin = JavaPlugin.getPlugin(BukkitPlatform.class);
110        this.bukkitWorld = Bukkit.getWorld(world.getName());
111        this.progressSubscribers.addAll(progressSubscribers);
112        this.forceSync = forceSync;
113        this.shouldGen = shouldGen;
114    }
115
116    @Override
117    public void start() {
118        if (!forceSync) {
119            // Request initial batch
120            this.requestBatch();
121            // Wait until next tick to give the chunks a chance to be loaded
122            TaskManager.runTaskLater(() -> task = TaskManager.runTaskRepeat(this, TaskTime.ticks(1)), TaskTime.ticks(1));
123        } else {
124            try {
125                while (!shouldCancel && !requestedChunks.isEmpty()) {
126                    chunkConsumer.accept(requestedChunks.poll());
127                }
128            } catch (Throwable t) {
129                throwableConsumer.accept(t);
130            } finally {
131                finish();
132            }
133        }
134    }
135
136    @Override
137    public void cancel() {
138        shouldCancel = true;
139    }
140
141    private void finish() {
142        try {
143            this.whenDone.run();
144        } catch (final Throwable throwable) {
145            this.throwableConsumer.accept(throwable);
146        } finally {
147            for (final ProgressSubscriber subscriber : this.progressSubscribers) {
148                subscriber.notifyEnd();
149            }
150            if (task != null) {
151                task.cancel();
152            }
153            finished = true;
154        }
155    }
156
157    @Override
158    public void run() {
159        if (shouldCancel) {
160            if (unloadAfter) {
161                Chunk chunk;
162                while ((chunk = availableChunks.poll()) != null) {
163                    freeChunk(chunk);
164                }
165            }
166            finish();
167            return;
168        }
169
170        Chunk chunk = this.availableChunks.poll();
171        if (chunk == null) {
172            if (this.availableChunks.isEmpty()) {
173                if (this.requestedChunks.isEmpty() && loadingChunks.get() == 0) {
174                    finish();
175                } else {
176                    requestBatch();
177                }
178            }
179            return;
180        }
181        long[] iterationTime = new long[2];
182        int processedChunks = 0;
183        do {
184            final long start = System.currentTimeMillis();
185            try {
186                this.chunkConsumer.accept(BlockVector2.at(chunk.getX(), chunk.getZ()));
187            } catch (final Throwable throwable) {
188                this.throwableConsumer.accept(throwable);
189            }
190            if (unloadAfter) {
191                this.freeChunk(chunk);
192            }
193            processedChunks++;
194            final long end = System.currentTimeMillis();
195            // Update iteration time
196            iterationTime[0] = iterationTime[1];
197            iterationTime[1] = end - start;
198        } while (iterationTime[0] + iterationTime[1] < this.maxIterationTime * 2 && (chunk = availableChunks.poll()) != null);
199        if (processedChunks < this.batchSize) {
200            // Adjust batch size based on the amount of processed chunks per tick
201            this.batchSize = processedChunks;
202        }
203
204        final int expected = this.expectedSize.addAndGet(-processedChunks);
205
206        if (expected <= 0) {
207            finish();
208        } else {
209            if (this.availableChunks.size() < processedChunks) {
210                final double progress = ((double) totalSize - (double) expected) / (double) totalSize;
211                for (final ProgressSubscriber subscriber : this.progressSubscribers) {
212                    subscriber.notifyProgress(this, progress);
213                }
214                this.requestBatch();
215            }
216        }
217    }
218
219    /**
220     * Requests a batch of chunks to be loaded
221     */
222    private void requestBatch() {
223        for (int i = 0; i < this.batchSize && this.requestedChunks.peek() != null; i++) {
224            // This required PaperLib to be bumped to version 1.0.4 to mark the request as urgent
225            final BlockVector2 chunk = this.requestedChunks.poll();
226            loadingChunks.incrementAndGet();
227            PaperLib
228                    .getChunkAtAsync(this.bukkitWorld, chunk.getX(), chunk.getZ(), shouldGen, true)
229                    .completeOnTimeout(null, 10L, TimeUnit.SECONDS)
230                    .whenComplete((chunkObject, throwable) -> {
231                        loadingChunks.decrementAndGet();
232                        if (throwable != null) {
233                            LOGGER.error("Failed to load chunk {}", chunk, throwable);
234                            // We want one less because this couldn't be processed
235                            this.expectedSize.decrementAndGet();
236                        } else if (chunkObject == null) {
237                            LOGGER.warn("Timed out awaiting chunk load {}", chunk);
238                            this.requestedChunks.offer(chunk);
239                        } else if (PlotSquared.get().isMainThread(Thread.currentThread())) {
240                            this.processChunk(chunkObject);
241                        } else {
242                            TaskManager.runTask(() -> this.processChunk(chunkObject));
243                        }
244                    });
245        }
246    }
247
248    /**
249     * Once a chunk has been loaded, process it (add a plugin ticket and add to
250     * available chunks list). It is important that this gets executed on the
251     * server's main thread.
252     */
253    private void processChunk(final @NonNull Chunk chunk) {
254        /* Chunk#isLoaded does not necessarily return true shortly after PaperLib#getChunkAtAsync completes, but the chunk is
255        still loaded.
256        if (!chunk.isLoaded()) {
257            throw new IllegalArgumentException(String.format("Chunk %d;%d is is not loaded", chunk.getX(), chunk.getZ());
258        }*/
259        if (finished) {
260            return;
261        }
262        chunk.addPluginChunkTicket(this.plugin);
263        this.availableChunks.add(chunk);
264    }
265
266    /**
267     * Once a chunk has been used, free it up for unload by removing the plugin ticket
268     */
269    private void freeChunk(final @NonNull Chunk chunk) {
270        if (!chunk.isLoaded()) {
271            throw new IllegalArgumentException(String.format("Chunk %d;%d is is not loaded", chunk.getX(), chunk.getZ()));
272        }
273        chunk.removePluginChunkTicket(this.plugin);
274    }
275
276    @Override
277    public int getRemainingChunks() {
278        return this.expectedSize.get();
279    }
280
281    @Override
282    public int getTotalChunks() {
283        return this.totalSize;
284    }
285
286    /**
287     * Subscribe to coordinator progress updates
288     *
289     * @param subscriber Subscriber
290     */
291    public void subscribeToProgress(final @NonNull ProgressSubscriber subscriber) {
292        this.progressSubscribers.add(subscriber);
293    }
294
295}