12
\$\begingroup\$

I'm trying to code following requirements with lockfree in most-used-code-path that is to get a resource from pool.

  1. Same resource should be used n (maxUsageCount) number of times and then destroyed and create a new one.

  2. If there is an error with a resource, it should be marked stale by client. Staled resource should be destroyed and not handed out to the clients, a new resource should be created.

  3. Resource must be closed after maxUsageCount otherwise its leak.

Resource is threadsafe.

public class ReusableResource<T extends ResourceProxy> {

    public static interface Factory<T extends ResourceProxy> {
        T create();
    }

    private final int maxUsageCount;
    private final AtomicReference<ConcurrentLinkedQueue<T>> resources;
    private final Object lock;
    private final ReusableResource.Factory<T> factory;

    public ReusableResource(int maxUsageCount,
            ReusableResource.Factory<T> factory) {
        if (maxUsageCount < 0)
            throw new IllegalArgumentException("maxUsageCount " + maxUsageCount);
        if (factory == null)
            throw new IllegalArgumentException("factory " + factory);

        this.maxUsageCount = maxUsageCount;
        resources = new AtomicReference<ConcurrentLinkedQueue<T>>(
                new ConcurrentLinkedQueue<T>());
        lock = new Object();
        this.factory = factory;
    }

    public T get() {
        final ConcurrentLinkedQueue<T> tmpq = resources.get(); // load once

        T res = tmpq.poll();

        if (res == null) {
            synchronized (lock) {
                res = tmpq.poll(); // double check
                // empty queue, first time or none left
                if (res == null) {
                    final ConcurrentLinkedQueue<T> newq = new ConcurrentLinkedQueue<T>(
                            Collections.nCopies(maxUsageCount, factory.create()));
                    // checkout the one we are returning
                    res = newq.poll();
                    // overwrite q object, do not modify existing
                    this.resources.set(newq);
                    // no stale check on newly created instance
                    return res; // <<- exit,
                }
            }
        }

        if (res.isStaled()) {
            synchronized (lock) {
                // once staled its going to remain staled
                // no need for double check

                // call destroy on all remaining elements
                // so that it can actually be closed
                // this might cause more destroy then checkouts
                // and so destroy must be idempotent
                res.destroy();

                while ((res = tmpq.poll()) != null)
                    res.destroy();

                // create new resources queue, same as first time
                final ConcurrentLinkedQueue<T> newq = new ConcurrentLinkedQueue<T>(
                        Collections.nCopies(maxUsageCount, factory.create()));
                // checkout the one we are returning
                res = newq.poll();
                // overwrite q object, do not modify existing
                this.resources.set(newq);
            }
        }

        return res;
    }
}

ResourceProxy.java

public abstract class ResourceProxy implements Disposable {
    private final AtomicInteger usageCount;
    private volatile boolean staled;

    public ResourceProxy(int maxUsageCount) {
        if (maxUsageCount < 0) throw new IllegalArgumentException();
        this.usageCount = new AtomicInteger(maxUsageCount);
        this.staled = false;
    }

    @Override
    public void destroy() {
        if (usageCount.decrementAndGet() == 0) {
            closeResource();
        }
    }

    @Override
    public void setStaled() {
        this.staled = true;
    }

    @Override
    public boolean isStaled() {
        return staled;
    }

    abstract protected void closeResource();

}

Example usage:

public class QueueConnectionProxy extends ResourceProxy implements
        QueueConnection {

    public static class Factory implements
            ReusableResource.Factory<QueueConnectionProxy> {
        private final int maxUsageCount;

        public Factory(int maxUsageCount) {
            if (maxUsageCount < 0) throw new IllegalArgumentException();
            this.maxUsageCount = maxUsageCount;
        }

        @Override
        public QueueConnectionProxy create() {
            return new QueueConnectionProxy(maxUsageCount);
        }
    }

    private final QueueConnection delegate;

    public QueueConnectionProxy(int maxUsageCount) {
        super(maxUsageCount);
        delegate = createDelegate();
    }

    private QueueConnection createDelegate() {
        // create QueueConnection
        return createdQueueConnection
    }

    @Override
    protected void closeResource() {
        try {
            if (delegate != null) delegate.close();
        } catch (JMSException e) {
            e.printStackTrace();
            // TODO: log JMSException
        }
    }

    @Override
    public void close() throws JMSException {
        destroy();
    }

    // other delegate methods
    //

}

To create an instance of ReusableResource:

ReusableResource<QueueConnectionProxy> reuse = 
  new ReusableResource<QueueConnectionProxy>(
    10, new QueueConnectionProxy.Factory(10));

Client code:

QueueConnection qc = null;
try {
    qc = reuse.get();
} finally {
    try { qc.close(); }
    catch (JMSException e) { e.printStackTrace(); }
}
\$\endgroup\$
5
\$\begingroup\$

Welcome to Code Review.

You have posted an very good quality post and that's why it takes so long to get an answer.

I have 2 minor remarks to make.

First :

if (maxUsageCount < 0) throw new IllegalArgumentException();

You do this check 3 times, I have no problem with that.
But what if your maxUsageCount is 0 ?
I think this is an unwanted situation so better to do :

if (maxUsageCount <= 0) throw new IllegalArgumentException();


Second :

The clientCode :

QueueConnection qc = null;
try {
     qc = reuse.get();
} finally {
    try { qc.close(); }
    catch (JMSException e) { e.printStackTrace(); }
}

You do not check if qc is null, what is possible when you get an exception. You have 2 options for this.

QueueConnection qc = null;
try {
     qc = reuse.get();
} finally {
    if (qc!=null) {
        try { qc.close(); }
        catch (JMSException e) { e.printStackTrace(); }
    }
}

edit :
This last one is wrong cause QueueConnection doesn't implement AutoClosable.

Or from java 7 you can use try-with-resource When you extend QueueConnection and implement the AutoClosable you can do the following :

try (QueueConnectionAutoClosable qc = reuse.get()) {
    //custom code what you need to do.
} 
\$\endgroup\$
3
\$\begingroup\$

In general it looks good, now going through the code to also see if it actually is good.

1. Exception messages

Your exception messages should be made clearer, an example of them is at the constructor of ReusableResource. If maxUsageCount < 0 or factory == null, we get echoed the value of the variable back, but no indication whatsoever to what is actually wrong with it. It would be good to see it included. Also maxUsageCount == 0 is most likely not wanted, is it? Lastly on this point, I would prefer this.factory = Objects.requireNonNull(factory) if you only use factory once, as the case is here. If it would be used multiple times, then there should be a simple Objects.requireNonNull(factory) check at the top of the method, to ensure the check only runs once.

2. Initialize objects as early as possible.

You are initializing both resources and lock in the constructor, and not at declaration time, which looks suspicious. They do not depend on either maxUsageCount or factory passed into the constructor. Hence I see no reason to initialize them late. This prevents duplicate code when adding more constructors.
This also holds for delegate in QueueConnectionProxy.

3. Try to use more meaningful variable names.

As a reader of the code, it is very hard to get the meaning of tmpq in ReusableResource.get(), after some thinking (which is a precious resource) I realised it was meant to be temporary queue, just name it tempQueue then.

4. Consider factory methods.

The lines:

ReusableResource<QueueConnectionProxy> reuse = 
  new ReusableResource<QueueConnectionProxy>(
    10, new QueueConnectionProxy.Factory(10));

are simply too long. Your code would benefit from giving ReusableResource a static method that creates a new ReusableResource, more over it seems that the 10 parameter is specified two times there, though I might be wrong. In either case, your static method would be called as ReusableResource.newInstance(10) or ReusableResource.newInstance(10, 10). This also hides the specification of the type parameter multiple times.

5. Use try-with-resources if possible

I do not know all places where try-with-resources is possible in your code after a brief look, but the following seems a good candidate:

QueueConnection qc = null;
try {
    qc = reuse.get();
} finally {
    try { qc.close(); }
    catch (JMSException e) { e.printStackTrace(); }
}

Which would become:

try (QueueConnection qc = reuse.get()) {
    //use
}

For this you might need to implement AutoClosable yourself, I cannot tell if that is possible.

That leads to the end of the review, unfortunately I am only able to touch the basic concepts and not the low level concurrency stuff as I consider myself not experienced enough to give advice upon that yet.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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