Source code

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.TimeoutException;
048import java.util.concurrent.atomic.AtomicInteger;
049import java.util.function.Consumer;
050
051/**
052 * Utility that allows for the loading and coordination of chunk actions
053 * <p>
054 * The coordinator takes in collection of chunk coordinates, loads them
055 * and allows the caller to specify a sink for the loaded chunks. The
056 * coordinator will prevent the chunks from being unloaded until the sink
057 * has fully consumed the chunk
058 * </p>
059 **/
060public final class BukkitChunkCoordinator extends ChunkCoordinator {
061
062    private static final Logger LOGGER = LogManager.getLogger("PlotSquared/" + BukkitChunkCoordinator.class.getSimpleName());
063
064    private final List<ProgressSubscriber> progressSubscribers = new LinkedList<>();
065
066    private final Queue<BlockVector2> requestedChunks;
067    private final Queue<Chunk> availableChunks;
068    private final long maxIterationTime;
069    private final Plugin plugin;
070    private final Consumer<BlockVector2> chunkConsumer;
071    private final org.bukkit.World bukkitWorld;
072    private final Runnable whenDone;
073    private final Consumer<Throwable> throwableConsumer;
074    private final boolean unloadAfter;
075    private final int totalSize;
076    private final AtomicInteger expectedSize;
077    private final AtomicInteger loadingChunks = new AtomicInteger();
078    private final boolean forceSync;
079    private final boolean shouldGen;
080
081    private int batchSize;
082    private PlotSquaredTask task;
083    private volatile boolean shouldCancel;
084    private boolean finished;
085
086    @Inject
087    private BukkitChunkCoordinator(
088            @Assisted final long maxIterationTime,
089            @Assisted final int initialBatchSize,
090            @Assisted final @NonNull Consumer<BlockVector2> chunkConsumer,
091            @Assisted final @NonNull World world,
092            @Assisted final @NonNull Collection<BlockVector2> requestedChunks,
093            @Assisted final @NonNull Runnable whenDone,
094            @Assisted final @NonNull Consumer<Throwable> throwableConsumer,
095            @Assisted("unloadAfter") final boolean unloadAfter,
096            @Assisted final @NonNull Collection<ProgressSubscriber> progressSubscribers,
097            @Assisted("forceSync") final boolean forceSync,
098            @Assisted("shouldGen") final boolean shouldGen
099    ) {
100        this.requestedChunks = new LinkedBlockingQueue<>(requestedChunks);
101        this.availableChunks = new LinkedBlockingQueue<>();
102        this.totalSize = requestedChunks.size();
103        this.expectedSize = new AtomicInteger(this.totalSize);
104        this.batchSize = initialBatchSize;
105        this.chunkConsumer = chunkConsumer;
106        this.maxIterationTime = maxIterationTime;
107        this.whenDone = whenDone;
108        this.throwableConsumer = throwableConsumer;
109        this.unloadAfter = unloadAfter;
110        this.plugin = JavaPlugin.getPlugin(BukkitPlatform.class);
111        this.bukkitWorld = Bukkit.getWorld(world.getName());
112        this.progressSubscribers.addAll(progressSubscribers);
113        this.forceSync = forceSync;
114        this.shouldGen = shouldGen;
115    }
116
117    @Override
118    public void start() {
119        if (!forceSync) {
120            // Request initial batch
121            this.requestBatch();
122            // Wait until next tick to give the chunks a chance to be loaded
123            TaskManager.runTaskLater(() -> task = TaskManager.runTaskRepeat(this, TaskTime.ticks(1)), TaskTime.ticks(1));
124        } else {
125            try {
126                while (!shouldCancel && !requestedChunks.isEmpty()) {
127                    chunkConsumer.accept(requestedChunks.poll());
128                }
129            } catch (Throwable t) {
130                throwableConsumer.accept(t);
131            } finally {
132                finish();
133            }
134        }
135    }
136
137    @Override
138    public void cancel() {
139        shouldCancel = true;
140    }
141
142    private void finish() {
143        try {
144            this.whenDone.run();
145        } catch (final Throwable throwable) {
146            this.throwableConsumer.accept(throwable);
147        } finally {
148            for (final ProgressSubscriber subscriber : this.progressSubscribers) {
149                subscriber.notifyEnd();
150            }
151            if (task != null) {
152                task.cancel();
153            }
154            finished = true;
155        }
156    }
157
158    @Override
159    public void run() {
160        if (shouldCancel) {
161            if (unloadAfter) {
162                Chunk chunk;
163                while ((chunk = availableChunks.poll()) != null) {
164                    freeChunk(chunk);
165                }
166            }
167            finish();
168            return;
169        }
170
171        Chunk chunk = this.availableChunks.poll();
172        if (chunk == null) {
173            if (this.availableChunks.isEmpty()) {
174                if (this.requestedChunks.isEmpty() && loadingChunks.get() == 0) {
175                    finish();
176                } else {
177                    requestBatch();
178                }
179            }
180            return;
181        }
182        long[] iterationTime = new long[2];
183        int processedChunks = 0;
184        do {
185            final long start = System.currentTimeMillis();
186            try {
187                this.chunkConsumer.accept(BlockVector2.at(chunk.getX(), chunk.getZ()));
188            } catch (final Throwable throwable) {
189                this.throwableConsumer.accept(throwable);
190            }
191            if (unloadAfter) {
192                this.freeChunk(chunk);
193            }
194            processedChunks++;
195            final long end = System.currentTimeMillis();
196            // Update iteration time
197            iterationTime[0] = iterationTime[1];
198            iterationTime[1] = end - start;
199        } while (iterationTime[0] + iterationTime[1] < this.maxIterationTime * 2 && (chunk = availableChunks.poll()) != null);
200        if (processedChunks < this.batchSize) {
201            // Adjust batch size based on the amount of processed chunks per tick
202            this.batchSize = processedChunks;
203        }
204
205        final int expected = this.expectedSize.addAndGet(-processedChunks);
206
207        if (expected <= 0) {
208            finish();
209        } else {
210            if (this.availableChunks.size() < processedChunks) {
211                final double progress = ((double) totalSize - (double) expected) / (double) totalSize;
212                for (final ProgressSubscriber subscriber : this.progressSubscribers) {
213                    subscriber.notifyProgress(this, progress);
214                }
215                this.requestBatch();
216            }
217        }
218    }
219
220    /**
221     * Requests a batch of chunks to be loaded
222     */
223    private void requestBatch() {
224        for (int i = 0; i < this.batchSize && this.requestedChunks.peek() != null; i++) {
225            // This required PaperLib to be bumped to version 1.0.4 to mark the request as urgent
226            final BlockVector2 chunk = this.requestedChunks.poll();
227            loadingChunks.incrementAndGet();
228            PaperLib
229                    .getChunkAtAsync(this.bukkitWorld, chunk.getX(), chunk.getZ(), shouldGen, true)
230                    .orTimeout(10L, TimeUnit.SECONDS)
231                    .whenComplete((chunkObject, throwable) -> {
232                        loadingChunks.decrementAndGet();
233                        if (throwable != null) {
234                            if (throwable instanceof TimeoutException) {
235                                LOGGER.warn("Timed out awaiting chunk load {}", chunk);
236                                this.requestedChunks.offer(chunk);
237                            } else {
238                                LOGGER.error("Failed to load chunk {}", chunk, throwable);
239                                // We want one less because this couldn't be processed
240                                this.expectedSize.decrementAndGet();
241                            }
242                        } else if (chunkObject == null) {
243                            if (shouldGen) {
244                                LOGGER.error("Null chunk returned for chunk at {}", chunk);
245                            }
246                        } else if (PlotSquared.get().isMainThread(Thread.currentThread())) {
247                            this.processChunk(chunkObject);
248                        } else {
249                            TaskManager.runTask(() -> this.processChunk(chunkObject));
250                        }
251                    });
252        }
253    }
254
255    /**
256     * Once a chunk has been loaded, process it (add a plugin ticket and add to
257     * available chunks list). It is important that this gets executed on the
258     * server's main thread.
259     */
260    private void processChunk(final @NonNull Chunk chunk) {
261        /* Chunk#isLoaded does not necessarily return true shortly after PaperLib#getChunkAtAsync completes, but the chunk is
262        still loaded.
263        if (!chunk.isLoaded()) {
264            throw new IllegalArgumentException(String.format("Chunk %d;%d is is not loaded", chunk.getX(), chunk.getZ());
265        }*/
266        if (finished) {
267            return;
268        }
269        chunk.addPluginChunkTicket(this.plugin);
270        this.availableChunks.add(chunk);
271    }
272
273    /**
274     * Once a chunk has been used, free it up for unload by removing the plugin ticket
275     */
276    private void freeChunk(final @NonNull Chunk chunk) {
277        if (!chunk.isLoaded()) {
278            throw new IllegalArgumentException(String.format("Chunk %d;%d is is not loaded", chunk.getX(), chunk.getZ()));
279        }
280        chunk.removePluginChunkTicket(this.plugin);
281    }
282
283    @Override
284    public int getRemainingChunks() {
285        return this.expectedSize.get();
286    }
287
288    @Override
289    public int getTotalChunks() {
290        return this.totalSize;
291    }
292
293    /**
294     * Subscribe to coordinator progress updates
295     *
296     * @param subscriber Subscriber
297     */
298    public void subscribeToProgress(final @NonNull ProgressSubscriber subscriber) {
299        this.progressSubscribers.add(subscriber);
300    }
301
302}