3
\$\begingroup\$

Is there any way to easily remove synchronization from this code, when there is a validation step before acquiring a lock?

I tried an AtomicReference<Pair<Integer, Double>> to hold the state. But that complicates the algorithm a lot because I have to use compareAndSet() in accept(), and then if that returns false, I have to redo the checks for isFinished(), isAboveDailyLimit(), aboveTradingLimit().

import wojtek.events.*;
import wojtek.tinytype.CounterParty;
import wojtek.tinytype.DailyLimit;
import wojtek.tinytype.TradeValue;
import wojtek.tinytype.TradingLimit;

import java.time.Clock;
import java.time.ZonedDateTime;

public class CheckTradeAgent {
    private final CreditCheckAPI creditCheckAPI;
    private final Clock clock;
    private final CounterParty counterParty;
    private final TradingLimit tradingLimit;
    private final DailyLimit dailyLimit;
    private final CheckTradeEventListener eventListener;

    private final Object limitAndFinishLock = new Object();
    private int utilisedDailyLimit;
    private boolean finished;

    public CheckTradeAgent(CreditCheckAPI creditCheckAPI, CheckTradeEventListener eventListener, Clock clock, CounterParty counterParty, TradingLimit tradingLimit, DailyLimit dailyLimit) {
        this.creditCheckAPI = creditCheckAPI;
        this.clock = clock;
        this.counterParty = counterParty;
        this.tradingLimit = tradingLimit;
        this.dailyLimit = dailyLimit;
        this.eventListener = eventListener;
    }

    public void handle(String counterParty, int tradeValue) {
        try {
            handle(new TradeSubmitted(new CounterParty(counterParty), new TradeValue(tradeValue), now()));
        } catch (RuntimeException e) {
            eventListener.onError(e);
        }
    }

    private synchronized void handle(TradeSubmitted trade) {
        if (isFinished()) {
            reject(trade);
            return;
        }

        if (wrongCounterParty(trade)) {
            skip(trade);
        } else {
            if (isAboveDailyLimit(trade)) {
                reject(trade);
            } else {
                if (aboveTradingLimit(trade)) {
                    creditCheck(trade);
                }
                accept(trade);
            }
        }
    }

    private void accept(TradeSubmitted submitted) {
        synchronized (limitAndFinishLock) {
            utilisedDailyLimit = utilisedDailyLimit + submitted.tradeValue.value;
            if (utilisedDailyLimit == dailyLimit.value) {
                this.finished = true;
            }
        }
        eventListener.accepted(new TradeAccepted(submitted, now()));
        if (isFinished()) {
            eventListener.onFinished();
        }
    }

    private void creditCheck(TradeSubmitted submitted) {
        try {
            creditCheckAPI.validate(submitted.counterParty.value, submitted.tradeValue.value);
        } catch (CheckException e) {
            eventListener.onError(e);
        }
    }

    private boolean aboveTradingLimit(TradeSubmitted submitted) {
        return submitted.tradeValue.value > tradingLimit.value;
    }

    private boolean isAboveDailyLimit(TradeSubmitted submitted) {
        return getUtilisedDailyLimit() + submitted.tradeValue.value > dailyLimit.value;
    }

    private boolean wrongCounterParty(TradeSubmitted submitted) {
        return !counterParty.equals(submitted.counterParty);
    }

    private void reject(TradeSubmitted submitted) {
        eventListener.rejected(new TradeRejectedByDailyLimit(submitted, now()));
    }

    private void skip(TradeSubmitted submitted) {
        eventListener.skipped(new TradeSkippedByWrongCounterParty(submitted, now()));
    }

    private ZonedDateTime now() {
        return ZonedDateTime.now(clock);
    }

    public int getUtilisedDailyLimit() {
        synchronized (limitAndFinishLock) {
            return utilisedDailyLimit;
        }
    }

    private boolean isFinished() {
        synchronized (limitAndFinishLock) {
            return finished;
        }
    }
}
\$\endgroup\$
2
  • \$\begingroup\$ I don't get what you want to do. Do you want to remove synchronization completely? Substitute it with some other kind of sync? If so Why? \$\endgroup\$
    – Cfontes
    Commented Jan 22, 2015 at 12:15
  • \$\begingroup\$ I am trying to speed things up. I would like to replace synchronization with CAS or Fetch-And-Add, or something along those lines. \$\endgroup\$
    – Wojtek
    Commented Jan 22, 2015 at 12:22

2 Answers 2

3
\$\begingroup\$

You may be able reduce the amount of synchronisation, but you cannot eliminate it entirely. CAS or Fetch-And-Add are lower level operations that are abstracted away by Java so you cannot use them directly.

Firstly, some observations: due to the mechanisms of how the lock is acquired, your synchronization of handle will cause a random caller to be selected when multiple are waiting. This could cause a situation where trades are handled out of order, which would be evidenced in the timestamps. If this isn't desired, you should use a queue to ensure trades are processed fairly (although what constitutes fair is a matter of debate).

You shouldn't use your CheckException as a convenient way to skip past a bunch of code. Have creditCheck(...) return a boolean and use proper flow control, otherwise your code appears to accept a trade even if the credit check fails.

Some of your preconditions to a trade may allow you to reject a trade before entering the synchronisation block. If the CounterParty doesn't match, you will always reject a trade. If a trade is above the daily limit, you will reject the trade outright. If a trade is above the trading limit, you can run the credit check without synchronising and reject it if it fails. It also appears that you only ever increment utilisedDailyLimit so if any trade's value would exceed the daily limit (and assuming trade values cannot be negative) you may reject it with an unsynchronised check against the daily limit.

Once you've checked all these preconditions, you have to worry about concurrency. You will have to check the daily limit again, but if this is done on a single-threaded consumer thread, you do not have to synchronise at all because there is no concurrent access and you know the value cannot change. With this approach, may also wish to submit the eventListener.accepted(...) call to a thread pool to execute separately as this may also slow down execution calling the listener inline.

A BlockingQueue would be ideal for this kind of design.

\$\endgroup\$
1
\$\begingroup\$

The first strange thing that stroke me is the existence of synchronisation with an explicit member-lock in the code that's already synchronised with an intrinsic lock.

Next is the question: access to which resources need to be actually synchronised? It looks like they are utilisedDailyLimit and finished as they are the only non final members, and in fact they need to be available for multiple threads at the same time - so guarding these members with an exclusive lock is too much of an overhead. Considering that finished is derived from utilisedDailyLimit it can be confidently thrown away. So, now when we got only one variable to watch and update, we can try to use CAS:

public class CheckTradeAgent {
    private final CreditCheckAPI creditCheckAPI;
    private final Clock clock;
    private final CounterParty counterParty;
    private final TradingLimit tradingLimit;
    private final DailyLimit dailyLimit;
    private final CheckTradeEventListener eventListener;

    private final AtomicInteger utilisedDailyLimit = new AtomicInteger();

    public CheckTradeAgent(CreditCheckAPI creditCheckAPI, CheckTradeEventListener eventListener, Clock clock, CounterParty counterParty, TradingLimit tradingLimit, DailyLimit dailyLimit) {
        this.creditCheckAPI = creditCheckAPI;
        this.clock = clock;
        this.counterParty = counterParty;
        this.tradingLimit = tradingLimit;
        this.dailyLimit = dailyLimit;
        this.eventListener = eventListener;
    }

    public void handle(String counterParty, int tradeValue) {
        try {
            handle(new TradeSubmitted(new CounterParty(counterParty), new TradeValue(tradeValue), now()));
        } catch (RuntimeException e) {
            eventListener.onError(e);
        }
    }

    public int getUtilisedDailyLimit() {
        return utilisedDailyLimit.get();
    }

    private void handle(TradeSubmitted trade) {
        if (!canBeUtilised(submitted, getUtilisedDailyLimit())) {
            reject(trade);
        } else if (wrongCounterParty(trade)) {
            skip(trade);
        } else {
            if (aboveTradingLimit(trade)) {
                creditCheck(trade);
            }
            if (utilise(submitted)) {
                accept(submitted);
            } else {
                reject(submitted);
            }
        }
    }

    private boolean utilise(TradeSubmitted submitted) {
        boolean accepted = false;
        int current;
        while (canBeUtilised(submitted, current = getUtilisedDailyLimit()) && !accepted) {
            accepted = utilisedDailyLimit.compareAndSet(current, current + submitted.tradeValue.value);
        }
        if (accepted && isFinished(getUtilisedDailyLimit())) {
            eventListener.onFinished();
        }
        return accepted;
    }

    private boolean canBeUtilised(TradeSubmitted submitted, int utilisedLimit) {
        return !isFinished(utilisedLimit) && !isAboveDailyLimit(submitted, utilisedLimit);
    }

    private boolean isFinished(int utilisedLimit) {
        return utilisedLimit >= dailyLimit.value;
    }

    private boolean isAboveDailyLimit(TradeSubmitted submitted, int utilisedLimit) {
        return utilisedLimit + submitted.tradeValue.value > dailyLimit.value;
    }

    private void creditCheck(TradeSubmitted submitted) {
        try {
            creditCheckAPI.validate(submitted.counterParty.value, submitted.tradeValue.value);
        } catch (CheckException e) {
            eventListener.onError(e);
        }
    }

    private boolean aboveTradingLimit(TradeSubmitted submitted) {
        return submitted.tradeValue.value > tradingLimit.value;
    }

    private boolean wrongCounterParty(TradeSubmitted submitted) {
        return !counterParty.equals(submitted.counterParty);
    }

    private void reject(TradeSubmitted submitted) {
        eventListener.rejected(new TradeRejectedByDailyLimit(submitted, now()));
    }

    private void skip(TradeSubmitted submitted) {
        eventListener.skipped(new TradeSkippedByWrongCounterParty(submitted, now()));
    }

    private void accept(TradeSubmitted submitted) {
        eventListener.accepted(new TradeAccepted(submitted, now()));        
    }

    private ZonedDateTime now() {
        return ZonedDateTime.now(clock);
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.