Is it possible to solve the multithreaded bank account problem fast without locks?

Question

I'm trying to solve the multithreaded bank account problem* without using locks but using multiversion concurrency control. It's working. It's just a bit slow. How can I speed it up?

(*) I have 5 users, each starting with 200 - each randomly withdrawing 100 and depositing 100 into another bank account owned by another user. I expect bank balances to total 1000 by the end of run. No money should be lost or created. This part works with my implementation below.

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;

public class ConcurrentWithdrawer {

    private Map<String, Integer> database = new HashMap<>();
    private int transactionCount = 0;
    private final List<Transaction> transactions = Collections.synchronizedList(new ArrayList<>());

    public static void main(String[] args) {
        try {
            new ConcurrentWithdrawer().run();
        } catch (ExecutionException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private static int getRandomNumberInRange(int min, int max) {

        if (min >= max) {
            throw new IllegalArgumentException("max must be greater than min");
        }

        Random r = new Random();
        return r.nextInt((max - min) + 1) + min;
    }

    public void run() throws ExecutionException, InterruptedException {
        int startAmount = 200;
        int numberAccounts = 5;
        int totalMoney = 0;
        for (int i = 0; i < numberAccounts; i++) {
            database.put(String.format("account%d", i), startAmount);
            totalMoney += startAmount;
        }


        ThreadPoolExecutor executor =
                (ThreadPoolExecutor) Executors.newFixedThreadPool(5);

        List<Future> futures = new ArrayList<Future>();
        for (int i = 0; i < 5; i++) {
            futures.add(executor.submit(new Callable<Integer>() {
                @Override
                public Integer call() {

                    for (int j = 0; j < 5; j++) {

                        Transaction transaction = beginTransaction(transactions, database);

                        transaction.read("fromBalance", "fromAccountName", (context) -> {
                            int fromAccount = getRandomNumberInRange(0, 4);
                            String fromAccountName = String.format("account%d", fromAccount);
                            return fromAccountName;
                        }).read("toBalance", "toAccountName", (context) -> {
                            int toAccount = getRandomNumberInRange(0, 4);
                            String toAccountName = String.format("account%d", toAccount);
                            while (toAccountName.equals(context.lookupName("fromAccountName"))) {
                                toAccount = getRandomNumberInRange(0, 4);
                                toAccountName = String.format("account%d", toAccount);
                            }
                            return toAccountName;
                        }).write("fromAccountName", (writeContext) -> {
                            int difference;
                            TransactionContext context = writeContext.context;
                            if (context.get("fromBalance") >= 100) {
                                difference = 100;
                            } else {
                                difference = 0;
                            }
                            context.write(writeContext.writeStep, "fromAccountName", context.get("fromBalance") - difference);
                            context.put("difference", difference);
                        }).write("toAccountName", (writeContext) -> {
                            TransactionContext context = writeContext.context;
                            context.write(writeContext.writeStep, "toAccountName", context.get("toBalance") + context.get("difference"));
                        }).commit();

                    }

                    int foundMoney = 0;
                    for (int j = 0; j < numberAccounts; j++) {
                        Integer foundMoney1;
                        String account = String.format("account%d", j);
                        foundMoney1 = database.get(account);

                        foundMoney += foundMoney1;
                    }

                    return foundMoney;
                }
            }));
        }

        List<Integer> monies = new ArrayList<>();
        for (Future f : futures) {
            int foundMoney = (Integer) f.get();
            monies.add(foundMoney);
        }
        System.out.println("Totals while running");
        for (Integer money : monies) {
            System.out.println(money);
        }
        System.out.println("Expected money");
        System.out.println(totalMoney);
        System.out.println("Final money");
        int foundMoney = 0;
        for (int j = 0; j < numberAccounts; j++) {
            Integer foundMoney1;

            foundMoney1 = database.get(String.format("account%d", j));
            System.out.println(String.format(String.format("account%d %d", j, foundMoney1)));
            foundMoney += foundMoney1;
        }
        System.out.println(foundMoney);
        executor.shutdown();
    }

    private Transaction beginTransaction(List<Transaction> transactions, Map<String, Integer> database) {
        transactionCount = transactionCount + 1;
        Transaction transaction = new Transaction(transactions, transactionCount, database);
        this.transactions.add(transaction);
        return transaction;
    }

    private class Transaction {
        public Long readTimestamp = 0L;
        public Long writeTimestamp = 0L;
        public List<String> readTargets = new ArrayList<>();
        private List<Transaction> transactions;
        private final int id;
        private Map<String, Integer> database;
        private List<TransactionStep> steps = new ArrayList<>();
        private TransactionContext transactionContext = new TransactionContext();
        private boolean active = true;
        private boolean cancel = false;
        private long transactionFinish;
        private long transactionStart;
        private int reread;
        private boolean valid;

        public Transaction(List<Transaction> transactions, int id, Map<String, Integer> database) {
            this.transactions = transactions;
            this.id = id;
            this.database = database;
        }


        public Transaction read(String field, String name, Function<TransactionContext, String> keyGetter) {
            ReadStep step = new ReadStep(this, field, keyGetter);
            steps.add(step);
            transactionContext.registerStep(name, step);
            return this;
        }

        public Transaction write(String fieldName, Consumer<WriteContext> writer) {
            steps.add(new WriteStep(this, fieldName, writer));
            return this;
        }

        public boolean invalid() {
            long largestWrite = 0L;
            long largestRead = 0L;
            List<Transaction> cloned = new ArrayList<>(transactions);
            cloned.sort(new Comparator<Transaction>() {
                @Override
                public int compare(Transaction o1, Transaction o2) {
                    return (int) (o1.transactionStart - o2.transactionStart);
                }
            });

            for (Transaction transaction : cloned) {
                ArrayList<TransactionStep> clonedSteps = new ArrayList<>(transaction.steps);
                for (TransactionStep step : clonedSteps) {
                    for (TransactionStep thisStep : steps) {
                        if (step instanceof ReadStep && thisStep instanceof ReadStep) {
                            ReadStep thisReadStep = (ReadStep) thisStep;
                            ReadStep readStep = (ReadStep) step;
                            if (thisReadStep.key.equals(readStep.key)) {
                                if (thisReadStep.timestamp > readStep.timestamp) {
                                    return true;
                                }
                            }
                        }
                        if (step instanceof WriteStep && thisStep instanceof WriteStep) {
                            WriteStep thisWriteStep = (WriteStep) thisStep;
                            WriteStep writeStep = (WriteStep) step;
                            if (thisWriteStep.timestamp > writeStep.timestamp) {
                                return true;
                            }
                        }
                    }
                }

            }

            return false;
        }

        public void commit() {
            boolean needsRunning = true;
            int retryCount = 0;
            transactionStart = System.nanoTime();

            while (needsRunning || invalid()) {
                readTimestamp = 0L;
                writeTimestamp = 0L;
                readTargets.clear();
                retryCount++;
                active = true;

                for (TransactionStep step : steps) {
                    step.run(transactionContext);
                }

                needsRunning = false;
                if (cancel) {
                    needsRunning = true;
                    cancel = false;
                }
            }


            System.out.println(String.format("Retry count was %d", retryCount));


            for (TransactionStep step : steps) {
                if (step instanceof ReadStep) {
                    String key = ((ReadStep) step).key;
                    Integer value = transactionContext.context.get(key);
                    database.put(key, value);
                }
            }


            transactions.remove(this);
            transactionFinish = System.nanoTime();

        }
    }

    private interface TransactionStep {
        TransactionContext run(TransactionContext context);
    }

    private class ReadStep implements TransactionStep {
        private final String field;
        private final Function<TransactionContext, String> keyGetter;
        private boolean activated;
        private String key;
        public long timestamp;
        Transaction transaction;

        public ReadStep(Transaction transaction, String field, Function keyGetter) {
            this.transaction = transaction;
            this.field = field;
            this.keyGetter = keyGetter;
            this.activated = false;
        }

        public TransactionContext run(TransactionContext context) {
            if (!activated) {
                key = (String) this.keyGetter.apply(context);
            }
            activated = true;
            timestamp = System.nanoTime();
            context.put(field, database.get(key));
            if (transaction.readTimestamp == 0L) {
                transaction.readTimestamp = timestamp;
            }
            transaction.readTargets.add(key);


            return context;
        }
    }

    private class TransactionContext {
        public final HashMap<String, Integer> context;
        private Map<String, ReadStep> readSteps = new HashMap<>();

        public TransactionContext() {
            this.context = new HashMap<>();
        }

        public void registerStep(String name, ReadStep readStep) {
            readSteps.put(name, readStep);
        }

        public void put(String field, Integer integer) {
            this.context.put(field, integer);
        }

        public String lookupName(String name) {
            return readSteps.get(name).key;
        }

        public void write(WriteStep writeStep, String name, Integer newValue) {
            String key = lookupName(name);
            writeStep.key = key;
            context.put(key, newValue);
        }

        public Integer get(String field) {
            return this.context.get(field);
        }
    }

    private class WriteStep implements TransactionStep {
        public String key;
        private boolean activated;
        private String fieldName;
        private final Consumer<WriteContext> writer;
        public long timestamp;
        Transaction transaction;

        public WriteStep(Transaction transaction, String fieldName, Consumer<WriteContext> writer) {
            this.transaction = transaction;
            this.fieldName = fieldName;
            this.writer = writer;
            activated = false;
        }

        @Override
        public TransactionContext run(TransactionContext context) {
            timestamp = System.nanoTime();
            transaction.writeTimestamp = timestamp;
            writer.accept(new WriteContext(this, context));
            activated = true;
            return context;
        }
    }

    private class WriteContext {
        private final WriteStep writeStep;
        private final TransactionContext context;

        public WriteContext(WriteStep writeStep, TransactionContext context) {
            this.writeStep = writeStep;
            this.context = context;
        }
    }
}

Output I receive:

Retry count was 4511
Retry count was 671
Retry count was 5956
Retry count was 140
Retry count was 3818
Retry count was 3102
Retry count was 34
Retry count was 580
Retry count was 106
Retry count was 46
Retry count was 22
Retry count was 11478
Retry count was 199
Retry count was 33
Retry count was 715
Retry count was 263
Retry count was 6186
Retry count was 6846
Retry count was 7012
Retry count was 301
Retry count was 93
Retry count was 148
Retry count was 11
Retry count was 355
Retry count was 7
Totals while running
1000
1000
1000
1000
1000
Expected money
1000
Final money
account0 200
account1 700
account2 100
account3 0
account4 0
1000

BUILD SUCCESSFUL in 515ms

How do I make it efficient? I'm sure Postgres doesn't let transactions run thousands of times before admitting them.

Someone said the code was obfuscated. This code reads a value (and records a read) like an SQL read statement. The code needs access to the name of the field being accessed as well as the actual value being accessed. This is why the code is written like this. The following code says: Read the field name generated by this function, store the name into fromAccountName and store the resulting value into fromBalance.

transaction.read("fromBalance", "fromAccountName", (context) -> {
                            int fromAccount = getRandomNumberInRange(0, 4);
                            String fromAccountName = String.format("account%d", fromAccount);
                            return fromAccountName;
                        })

Answer 1

Short description of the system being tested, deducted from the question and looking at the code:

• Accounts are kept in a "database" that has that consists of account names mapped to account values.
• Transactions between accounts should always be atomic in nature, so that no money but the initial money of 200 is created or destroyed.
• An additional assumption is that accounts cannot have a negative value.
• In your example code, the database is implemented as a Map<String,Integer>.

The problem transfers being atomic can be solved by using an account database BankAccounts as follows. This excludes the test code, but solves the problem of consistency.

public class BankAccounts {
    /**
     * The number of retries for each transfer
     */
    public static final int RETRIES = 10;
    /**
     * The account database
     */
    private Map<String, AtomicInteger> accounts = new HashMap<>();

    /**
     * Creates accounts with each 200 initial balance.
     */
    public BankAccounts() {
        // fill the accounts initially
        // Left as an exercice to the reader
    }

    /**
     * Return a set with all account names.
     */
    public Set<String> accountNames() {
        return Collections.unmodifiableSet(accounts.keySet());
    }

    /**
     * Get the balance value for the specified account.
     */
    public int getBalanceFor(String accountName) {
        AtomicInteger account = accounts.get(accountName);
        return account != null ? account.get() : 0;
    }

    /**
     * Atomically transfers an amount from one account to another and returns {@code true} if that was successful.
     */
    public boolean transfer(String fromAccountName, String toAccountName, int amount) {
        AtomicInteger fromBalance = accounts.get(fromAccountName);
        AtomicInteger toBalance = accounts.get(toAccountName);
        if (amount < 0 || fromBalance == null || toBalance == null) {
            return false;
        }

        for (int retry = 0; retry < RETRIES; retry++) {
            int fromValue = fromBalance.get();  // get from-account balance value
            if (fromValue >= amount) { // check if enough money
                if (fromBalance.compareAndSet(fromValue, fromValue - amount)) {
                    toBalance.addAndGet(amount); // Adding money is always allowed ;-) 
                    return true;
                } else {
                    // value of fromBalance was changed concurrently
                    Thread.yield(); // optional.
                }
            }
        }
        return false;
    }
}