Cfrtjryk

I've been told that the following code has numerous synchronization problems, but I don't have sufficient experience with multithreading to identify and resolve them. Any feedback would be great.

I have a number of asynchronous threads that want to access a database for reads and writes. The rate of collisions is very high, and there are numerous instances where one thread's operation depends on the completion of another thread's operation, so I implemented a thread with a queue to synchronize these accesses.

Every task is an implementation of DBRequest. The actual content varies, but the result of all activities is a string. An asynchronous thread creates a DBRequest and submits it to the queue. The queue runs the task, which produces a string. How do I get that string back to the thread that created the DBRequest, and how can I cause my creator thread to wait for the result?

public interface DBRequest {

    String execute(VdtsSysDB vdtsSysDB, BoardLoad currentLoad);

}



public class DBQueue implements Runnable {

    private static DBQueue dbQueue;

    private LinkedBlockingQueue<DBRequest> queue = new LinkedBlockingQueue<>();

    private VdtsSysDB vdtsSysDB = new VdtsSysDB();

    private ReentrantLock lock = new ReentrantLock();

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

    private boolean kill = false;



    private BoardLoad currentLoad;

    private ProgressController progressController;



    public static DBQueue getInstance() {

        if (dbQueue == null) synchronized (DBQueue.class) {

            if (dbQueue == null)

                dbQueue = new DBQueue();

        }

        return dbQueue;

    }



    private DBQueue() {

    }



    public ReentrantLock getLock() {

        return lock;

    }



    @Override

    public void run() {

        LOG.info("Starting DBQueue loop. Kill {}.", kill);

        while (!kill) {

            DBRequest dbRequest = removeRequest();

            if (dbRequest != null) {

                lock.lock();

                String result = dbRequest.execute(vdtsSysDB, currentLoad);

                lock.unlock();

                if (progressController != null) Platform.runLater(() ->

                        progressController.updateDisplay(currentLoad));

            }

        }

        vdtsSysDB.getEntityManager().close();

    }





    public void addRequest(DBRequest dbRequest) {

        try {

            queue.add(dbRequest);

            LOG.info("Added request.");

        } catch (Exception e) {

            LOG.error("Can't add element.", e);

        }

    }



    private DBRequest removeRequest() {

        DBRequest result = null;

        try {

            //result = queue.poll(10, TimeUnit.SECONDS);

            result = queue.take();

        } catch (Exception e) {

            LOG.error("Exception.", e);

        }

        return result;

    }



    public void killDBQueue() {

        kill = true;

        LOG.info("Shutting down DBQueue.");

    }



    public static void start() {

        Thread thread = new Thread(DBQueue.getInstance(), "DBQueue Thread");

        thread.start();

        LOG.info("Starting DBQueue.");

    }



    public BoardLoad getCurrentLoad() {

        if (currentLoad == null)

            currentLoad = BoardLoad.getLastOpenLoad(vdtsSysDB);

        return currentLoad;

    }



    public void setCurrentLoad(BoardLoad proposedLoad) {

        // We can only have one open load, and by definition, the current load is open. So close it.

        if (this.currentLoad != null && !this.currentLoad.equals(proposedLoad)) {

            currentLoad.close(vdtsSysDB);

            if (proposedLoad != null) {

                this.currentLoad = vdtsSysDB.getEntityManager().find(BoardLoad.class, proposedLoad.getId());

            } else this.currentLoad = null;

        }

    }



    public ProgressController getProgressController() {

        return progressController;

    }



    public void setProgressController(ProgressController progressController) {

        this.progressController = progressController;

    }

}