Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

I have implemented a ThreadPoolExecutor that will run a Consumer<T> only on elements not already consumed. This code uses Java 8.

The background behind this is that I scan a directory every x time units for which files are present, I must maintain a 100% accuracy on finding files and other mechanisms such as JNotify or a simple plain WatcherService do not achieve that.

public class SingleExecutionThreadPoolExecutor<E> extends ThreadPoolExecutor {
    private final Consumer<E> consumer;
    private final List<E> elementsInProcess = new ArrayList<>();

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Consumer<E> consumer) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        this.consumer = consumer;
    }

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Consumer<E> consumer, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        this.consumer = consumer;
    }

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Consumer<E> consumer, ThreadFactory threadFactory) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        this.consumer = consumer;
    }

    public SingleExecutionThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Consumer<E> consumer, ThreadFactory threadFactory, RejectedExecutionHandler handler) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        this.consumer = consumer;
    }

    public void execute(E element) {
        if (!elementsInProcess.contains(element)) {
            super.execute(() -> consumer.accept(element));
            elementsInProcess.add(element);
        }
    }

    @Override
    @Deprecated
    public void execute(Runnable command) { }
}

It is called using the following classes and snippets:

public final class UniqueTimePath {
    private final Path path;
    private final FileTime fileTime;

    public UniqueTimePath(final Path path) {
        this.path = path;
        try {
            this.fileTime = Files.getLastModifiedTime(path);
        } catch (IOException ex) {
            throw new UncheckedIOException(ex);
        }
    }

    public Path getPath() {
        return path;
    }

    public FileTime getFileTime() {
        return fileTime;
    }

    @Override
    public int hashCode() {
        int hash = 3;
        return hash;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final UniqueTimePath other = (UniqueTimePath) obj;
        if (!Objects.equals(this.path, other.path)) {
            return false;
        }
        if (!Objects.equals(this.fileTime, other.fileTime)) {
            return false;
        }
        return true;
    }

    @Override
    public String toString() {
        return "{" + path + ", " + fileTime + "}";
    }
}

private SingleExecutionThreadPoolExecutor<UniqueTimePath> executor;
//...
executor = new SingleExecutionThreadPoolExecutor<>(threadCount, threadCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(), fileConsumer);
//...
Files.list(directory)
        .map(UniqueTimePath::new)
        .forEach(executor::execute);
share|improve this question
add comment

3 Answers 3

up vote 10 down vote accepted

.equals

The last part of your .equals implementation can be simplified to:

return Objects.equals(this.path, other.path) && Objects.equals(this.fileTime, other.fileTime);

HashCode

int hash = 3;
return hash;

Do I even have to comment anything about what I think about this implementation? You use Objects.equals above, you can use Objects.hashCode(this.path, this.fileTime); here. It is preferred if hashCode and equals use the same fields for their operations.

Set or List?

Whenever you use a List in a program, ask yourself the question: Does the order of the elements matter?

In this case, they don't. Use the Set interface and create as HashSet for your elementsInProcess variable.

private final Set<E> elementsInProcess = new HashSet<>();

When using a Set, it is extra important that the elements in the set has properly implemented hashCode and equals! (Or are using the default implementations inherited from the Object class)

Cleanup

I suggest you override the shutdown methods of a ThreadPoolExecutor to clear the elementsInProcess collection.

execute(Runnable command)

Please throw a UnsupportedOperationException or similar in this method:

public void execute(Runnable command) { }

Edit: Scratch that. I think palacsint is absolutely right. By doing nothing here, you're breaking the contract of the Executor interface (which you implement, whether you know it or not - by extending an ExecutorService). The Javadoc for the execute(Runnable command) method says:

Executes the given command at some time in the future.

As it currently stands, you are not doing that and have no real ability of doing so in that method. This fact is an indication that you should not use extends ThreadPoolExecutor and instead use a private ThreadPoolExecutor executor;, and therefore using composition over inheritance.

Concurrency

If two threads would call execute at the same time, there is a concurrency issue on elementsInProcess. You need to synchronize on something there, for minimal blocking I suggest you do something like this:

synchronize (lock) {
    if (elementsInProcess.contains(element)) {
        return;
    }
    elementsInProcess.add(element);
}
super.execute(() -> consumer.accept(element));

Where lock is a private final Object lock = new Object();

Or, you could utilize the fact that the .add method of a set actually returns a boolean indicating if the add was successful and use a Collections.synchronizedSet.

if (!elementsInProcess.add(element)) {
    return;
}
super.execute(() -> consumer.accept(element));

It is important when using a synchronized Set that you only do one method call here, as there could otherwise be a gap between .contains and .add which would again cause concurrency issues between threads.

share|improve this answer
add comment
  1. I would definitely use composition here instead of inheritance. It would eliminate the following ugly (and smelly) code too:

    @Override
    @Deprecated
    public void execute(Runnable command) { }
    

    The internals of ThreadPoolExecutor could be changed by JDK developers, so in a future release submit() methods might not call the overridden empty execute() method any more or they could provide a new public execute method which bypass your execute().

    See also: Effective Java, Second Edition, Item 16: Favor composition over inheritance

  2. The code should check that the consumer is null in the constructors and throw an IllegalArgumentException or NullPointerException right there. If it remains null you'll get an exception later in the execute method anyway. Throwing an exception immediately helps debugging a lot since you get a stacktrace with the frames of the erroneous client, not just a NullPointerException later, probably from another thread. (The Pragmatic Programmer: From Journeyman to Master by Andrew Hunt and David Thomas: Dead Programs Tell No Lies.)

  3. Most developers would expect that submitting jobs from different threads is thread-safe, so I'd use a properly synchronized elementsInProcess list instead of ArrayList to meet this expectation.

  4. The following will lead to poor performance if you use your objects in a HashMap:

    @Override
    public int hashCode() {
        int hash = 3;
        return hash;
    }
    

    Consider implementing a proper hashCode.

share|improve this answer
    
Is composition here worth the fact that you need to add delegating methods for all functionality of ThreadPoolExecutor? –  skiwi Apr 3 at 10:02
1  
@skiwi: Eclipse can generate those methods for you, so it is not too much work but you will have a lot more control, better API and less bugs in the long run. The referred Effective Java item mentions the Properties class as famous bad example, read that chapter if you can. (stackoverflow.com/a/11344295/843804) –  palacsint Apr 3 at 10:07
2  
If new methods are added, you should consider which must be available from your class. Allowing a helper class to dictate your class's API without your knowledge or ability to vet the changes is a recipe for disaster, i.e. random bugs. –  David Harkness Apr 4 at 2:25
1  
+1 to composition over inheritance for this one. I'm not usually a strong advocate for this principle, but I am a strong advocate for the Liskov Substitution Principle (see oodesign.com/liskov-s-substitution-principle.html), which your code violates. Using composition and not exposing the parts of the API that you prevent from operating as their contract specifies would be the best approach here. (Yes, it's also violated by several parts of the core Java library, but that is the most serious design error in Java to my mind.) –  Jules Apr 4 at 4:51
1  
@skiwi Do you really need to expose all the methods of a ThreadPoolExecutor? Your example of use only uses a small handful of them, so why not only expose those? –  Jules Apr 4 at 4:55
show 2 more comments

I agree with @SimonAndréForsberg and @palacsint's answers, I have however reconsidered the design and noticed that I do not even want to really use composition for the ThreadPoolExecutor.

I have refactored the code to the class below, where the following assumptions hold:

  • The Consumer<T> is already known.
  • The Executor has already been instantiated, in this case an instantiation of a ThreadPoolExecutor.
public class SingleExecutionExecutorBridge<T> implements Consumer<T> {
    private final Consumer<T> consumer;
    private final Executor executor;

    private final Set<T> elementsInProcess = Collections.synchronizedSet(new HashSet<>());

    public SingleExecutionExecutorBridge(final Consumer<T> consumer, final Executor executor) {
        this.consumer = Objects.requireNonNull(consumer);
        this.executor = Objects.requireNonNull(executor);
    }

    @Override
    public void accept(final T element) {
        if (!elementsInProcess.add(element)) {
            return;
        }
        executor.execute(() -> consumer.accept(element));
    }
}

With this approach everything is decoupled nicely, and the following code snippets are now relevant:

private final SingleExecutionExecutorBridge<UniqueTimePath> singleExecutionExecutorBridge;
private ThreadPoolExecutor executor;

//...

public BaseChecker(final Path directory, final Consumer<UniqueTimePath> fileConsumer, final Path configFile) {
    this.directory = Objects.requireNonNull(directory);
    this.configFile = Objects.requireNonNull(configFile);
    this.subClass = this.getClass();
    this.singleExecutionExecutorBridge = new SingleExecutionExecutorBridge<>(Objects.requireNonNull(fileConsumer), executor);
}

//...

executor = new ThreadPoolExecutor(threadCount, threadCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>());

//...

Files.list(directory)
        .map(UniqueTimePath::new)
        .forEach(singleExecutionExecutorBridge::accept);
share|improve this answer
    
Nice, well done. You might want to add Objects.requireNonNull in your constructor for SingleExecutionExecutorBridge though –  Simon André Forsberg Apr 3 at 12:18
    
@SimonAndréForsberg Yes, yes I do, I'll edit the post to include them. –  skiwi Apr 3 at 12:19
3  
I have to say that what you did here is composition :-). Unfortunately I didn't have time earlier to show a complete example like this one but I'm glad that you write it. Furthermore, I've found the constructor injection really easy to test with mocked executors. So it's much better than creating a new ThreadPoolExecutor inside the class could be. –  palacsint Apr 3 at 23:09
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

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