Блог

Я работаю над игрой, и у меня проблема с синхронизацией потоков. Мне с треском не удалось написать это правильно, поэтому основной поток не зависает из-за этого. Короткая история о создании блокировок в моем основном игровом потоке (игровом цикле). Я сделаю все возможное, чтобы предоставить детали, насколько смогу. Проблема не связана с фреймворком игры, поскольку это мой общий код, который блокирует все.

Предыстория:

Действия, которые происходят в игре, выполняются как сопрограммы. Поскольку в Java нет функции сопрограммы, они были реализованы как сопрограммы с использованием потоков. Идея состоит в том, чтобы иметь прерываемые действия, поэтому при запуске нового действия выполнение текущего приостанавливается до завершения нового. Это поведение также может иметь большую глубину.

Вопрос:

Как правильно выполнить синхронизацию в этом случае или как правильно защитить мой цикл основного потока (метод обновления, приведенный в примере), чтобы он не зависал?

Вот упрощенный вариант использования. Каждое выполняемое действие, т.Е. Перемещение единиц, Обновление всего, что было выполнено, выполнялось как сопрограммы или набор сопрограмм.

 Game init:
Player turn: move units around, upgrade stuff and so on
Ai turn:
    Calculate stuff
    Calculate more stuff
    Evaluate player strength
    Evaluate choke points
    More calculations
    Plan attack actions
    Move units>
        Create action (Coroutines) for every single move >
            Coroutine executing for move
                We have to fight in battle between two units
Create action (Coroutines) for every Combat
                        Coroutines executing for Combat
                       Coroutine for combat finished
            Coroutine for move finished
            repeat few times, sometime with and without battle
        Move Coroutine finished
    Move loop finished
Do more stuff
End ai turn;
  

Я просматривал дампы потоков, чтобы выяснить, что является причиной этого, и его метод возобновления.

Дампы потоков, где вы можете видеть блокировки:

ОСНОВНОЙ поток заблокирован :

 
"LWJGL Application" #18 prio=5 os_prio=0 tid=0x0000000020062800 nid=0x3b20 in Object.wait() [0x0000000022a7f000]
   java.lang.Thread.State: WAITING (on object monitor)
        at java.lang.Object.wait(Native Method)
        at java.lang.Object.wait(Object.java:502)
        at package_name.Coroutine$CoroutineExecutor.resume(Coroutine.java:319)
        - locked <0x00000006c3aeddd8> (a java.lang.Thread)
        at package_name.Coroutine.resume(Coroutine.java:409)
        at package_name.screens.GameScreen.render(GameScreen.java:430)
        at com.badlogic.gdx.Game.render(Game.java:46)
        at package_name.Game.render(Game.java:273)
        at com.badlogic.gdx.backends.lwjgl.LwjglApplication.mainLoop(LwjglApplication.java:223)
        at com.badlogic.gdx.backends.lwjgl.LwjglApplication$1.run(LwjglApplication.java:124)

   Locked ownable synchronizers:
        - None
  

ДРУГОЙ поток заблокировал ОСНОВНОЙ поток:

 "Thread-6" #32 prio=5 os_prio=0 tid=0x0000000020a9e000 nid=0xc94 runnable [0x000000014d64e000]
   java.lang.Thread.State: RUNNABLE
        //Doing something here which might take a second or two
        at package_name.Coroutine$CoroutineExecutor$1.run(Coroutine.java:242)
        - locked <0x00000006c3aeddd8> (a java.lang.Thread)
        at java.lang.Thread.run(Thread.java:748)

   Locked ownable synchronizers:
        - None
  

Полный код сопрограммы:

 package com.game.coroutine;

import com.game.coroutine.CoroutineDeath;
import com.game.coroutine.DeadCoroutineException;
import com.game.coroutine.InternalCoroutineException;
import com.game.coroutine.ResumeSelfCoroutineException;
import com.game.coroutine.YieldOutsideOfCoroutineException;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * An implementation of Lua-like coroutines.
 * Supports returning values with yield, nesting and it might be thread safe.
 */
public class Coroutine {
    private static final Logger log = LoggerFactory.getLogger(Coroutine.class);
    private static List<CoroutineExecutor> executorPool = new ArrayList<>();

    private final Runnable runnable;
    private Object returnValue;
    private boolean finished;

    /** I think CoroutineExecutor is used to map Lua coroutines (not present in Java) to Java Threads.
     * 
     * */
    private static class CoroutineExecutor {

        private boolean running = true;
        private final Thread thread;
        private Coroutine coroutine;

        CoroutineExecutor() {
            thread = new Thread(new Runnable() {
                @Override
                public void run() {
                    synchronized (thread) {
                        while (running) {
                            try {
                                if (!coroutine.finished) {
                                    coroutine.runnable.run();
                                }
                            } catch (CoroutineDeath | Exception e) {
                                log.error("Error while running coroutine runnable", e);
                            }
                            finally {
                                coroutine.finished = true;
                                coroutine.returnValue = null;
                                thread.notifyAll();
                                try {
                                    thread.wait();
                                } catch (InterruptedException e) {
                                    log.error("Error while waiting for coroutine runnable", e);
                                    Thread.currentThread().interrupt();
                                }
                            }
                        }
                    }
                }
            });
            thread.setDaemon(true);
            coroutine = new Coroutine(new Runnable() {
                @Override
                public void run() { }
            });
            coroutine.finished = true;
        }

        Object resume() {
            synchronized (thread) {
                if (thread.getState() == Thread.State.NEW) {
                    thread.start();
                }
                thread.notifyAll();
                try {
                    thread.wait();
                } catch (InterruptedException e) {
                    throw new InternalCoroutineException("Thread was interrupted while waiting for coroutine to yield.");
                }
                return coroutine.returnValue;
            }
        }

        void yield(Object o) {
            synchronized (thread) {
                coroutine.returnValue = o;
                thread.notifyAll();
                try {
                    thread.wait();
                } catch (InterruptedException interrupted) {
                    throw new CoroutineDeath();
                }
            }
        }

        void setCoroutine(Coroutine coroutine) {
            synchronized (thread) {
                if (busy()) {
                    throw new InternalCoroutineException("Coroutine assigned to a busy executor.");
                }
                this.coroutine = coroutine;
            }
        }

        boolean busy() {
            synchronized (Coroutine.class) {
                return !coroutine.finished;
            }
        }

        void reset() {
            synchronized (thread) {
                coroutine.finished = true;
                if (thread.getState() != Thread.State.NEW) {
                    thread.interrupt();
                    try {
                        thread.wait();
                    } catch (InterruptedException e) {
                        log.error("Error while waiting for coroutine runnable", e);
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }

        void kill() {
            synchronized (thread) {
                running = false;
                thread.interrupt();
                try {
                    thread.join();
                } catch (InterruptedException e) {
                    log.error("Error while waiting for coroutine runnable", e);
                    Thread.currentThread().interrupt();
                }
            }
        }

        @Override
        public String toString() {
            return "Executor "   thread.getName();
        }
    }

    private Coroutine(Runnable runnable) {
        this.runnable = runnable;
        this.finished = false;
    }


    /**
     * Resumes/starts a coroutine. Return value is the object that the coroutine passed to yield
     */
    public static <T> T resume(Coroutine coroutine) {

        if (coroutine.finished) {
            throw new DeadCoroutineException("An attempt was made to resume a dead coroutine.");
        }

        CoroutineExecutor executor = getExecutorForCoroutine(coroutine);

        if(executor != null) {
            if (executor.equals(getCurrentExecutor())) {
                throw new ResumeSelfCoroutineException("A coroutine cannot resume itself.");
            }
            return (T) executor.resume();
        }
        else {
            log.error("CoroutineExcutor is null");
            return null;
        }
    }


    /**
     * A coroutine can use this to return a value to whatever called resume
     */
    public static void yield(Object o) {

        CoroutineExecutor coroutineExecutor = getCurrentExecutor();
        if (coroutineExecutor != null) {
            Coroutine coroutine = coroutineExecutor.coroutine;

            if (coroutine == null) {
                throw new YieldOutsideOfCoroutineException("Yield cannot be used outside of a coroutine.");
            }

            coroutineExecutor.yield(o);
        } else {
            log.error("CoroutineExcutor is null");
        }

    }


    /**
     * Creates a new coroutine that with the "body" of runnable, doesn't start until resume is used
     */
    public static synchronized  Coroutine create(Runnable runnable) {
        Coroutine coroutine = new Coroutine(runnable);
        CoroutineExecutor coroutineExecutor = getFreeExecutor();
        coroutineExecutor.setCoroutine(coroutine);

        return coroutine;
    }


    /**
     * Stops and cleans up the coroutine
     */
    public static synchronized void destroy(Coroutine coroutine) {
        CoroutineExecutor executor = getExecutorForCoroutine(coroutine);
        if (executor != null) {
            executor.reset();
        }
    }

    /**
     * Returns true if resuming this coroutine is possible
     */
    public static synchronized boolean alive(Coroutine coroutine) {
        return coroutine != null amp;amp; !coroutine.finished;
    }

    /**
     * Shrinks the thread pool
     */
    public static synchronized void cleanup() {
        Iterator<CoroutineExecutor> it = executorPool.iterator();
        while (it.hasNext()) {
            CoroutineExecutor executor = it.next();
            if (!executor.busy()) {
                executor.kill();
                it.remove();
            }

        }
    }

    /**
     * Returns the current number of executors in the pool
     */
    public static synchronized int poolSize() {
        return executorPool.size();
    }

    private static synchronized CoroutineExecutor getCurrentExecutor() {
        for (CoroutineExecutor e : executorPool) {
            if (Thread.currentThread().equals(e.thread)) {
                return e;
            }
        }
        return null;
    }

    private static synchronized CoroutineExecutor getFreeExecutor() {
        for (CoroutineExecutor executor : executorPool) {
            if (!executor.busy()) {
                return executor;
            }
        }
        CoroutineExecutor newExecutor = new CoroutineExecutor();
        executorPool.add(newExecutor);
        return newExecutor;
    }

    private static synchronized CoroutineExecutor getExecutorForCoroutine(Coroutine coroutine) {
        for (CoroutineExecutor executor : executorPool) {
            if (coroutine.equals(executor.coroutine)) {
                return executor;
            }
        }
        return null;
    }
}
  

Наконец, игровой цикл в основном потоке, который застревает из-за возобновления, которое заблокировало объект потока сопрограммы:

 public boolean update() {
    Coroutine coroutine = coroutineQueue.peek();
    if (coroutine != null) {
        if (Coroutine.alive(coroutine)) {
            Event event = Coroutine.resume(coroutine);
            if (event != null) {
                broadcast(event);
            }
        } else {
            coroutineQueue.poll();
        }
    }
    return !coroutineQueue.isEmpty();
}
  

Пожалуйста, посоветуйте, как исправить синхронизацию, чтобы в конце дня основной цикл не блокировался, а все остальные сопрограммы выполнялись правильно, последовательно и при необходимости приостанавливались / продолжались.

Спасибо всем, что нашли время, чтобы прочитать этот вопрос. С уважением

Если вы не предоставляете блокировку для synchronized ключевого слова, по умолчанию используется объект. Но в вашем случае все это статические методы, поэтому в качестве блокировки используется класс. Итак, все ваши методы сопрограммы блокируют одно и то же — класс сопрограммы.