1
\$\begingroup\$

I am working on a project in which I will be having different Bundles. Let's take an example, Suppose I have 5 Bundles and each of those bundles will have a method name process.

Below are the things, I am supposed to do-

  1. I need to call all those 5 Bundles process method in parallel using multithreaded code and then write to the database. Meaning each bundle will return me back Map of String and String and then I am supposed to write that map into the database. I am not sure what is the right way to do that? Should I have five thread? One thread for each bundle? But what will happen in that scenario, suppose if I have 50 bundles, then I will have 50 threads?
  2. And also, I want to have timeout feature as well. If any bundles is taking lot of time than the threshold setup by us, then it should get timeout and log as an error that this bundle has taken lot of time.

The following attempt that I have done is most probably flawed and error handling is by no means complete. And I am looking for best and efficient way of doing this problem.

Below is the solution I have. And below is my method which will call process method of all the bundles in a multithreaded way.

public void processingEvents(final Map<String, Object> eventData) throws InterruptedException {
    ExecutorService pool = Executors.newFixedThreadPool(5);
    List<ProcessBundleHolderEntry> entries = new ArrayList<ProcessBundleHolderEntry>();

    Map<String, String> outputs = (Map<String, String>)eventData.get(BConstants.EVENT_HOLDER);

    for (BundleRegistration.BundlesHolderEntry entry : BundleRegistration.getInstance()) {
        ProcessBundleHolderEntry processBundleHolderEntry = new ProcessBundleHolderEntry(entry, outputs);
        entries.add(processBundleHolderEntry);
    }

    try {
        List<Future<Map<String, String>>> futures = pool.invokeAll(entries, 30L, TimeUnit.SECONDS);
        for (int i = 0; i < futures.size(); i++) {
            // This works since the list of future objects are in the
            // same sequential order as the list of entries
            Future<Map<String, String>> future = futures.get(i);
            ProcessBundleHolderEntry entry = entries.get(i);
            if (!future.isDone()) {
                // log error for this entry
            }
        }
    } catch (InterruptedException e) {
        pool.shutdownNow(); //cancels the tasks
        //restore interrupted flag and exit
        Thread.currentThread().interrupt();
        //or rethrow the exception
        throw e;
    }
}

Secondly, an implementation of Callable for your threads:

public class ProcessBundleHolderEntry implements Callable<Map<String, String>> {
    private BundleRegistration.BundlesHolderEntry entry;
    private Map<String, String> outputs;

    public ProcessBundleHolderEntry(BundleRegistration.BundlesHolderEntry entry, Map<String, String> outputs) {
        this.entry = entry;
        this.outputs = outputs;
    }

    public Map<String, String> call() throws Exception {
        final Map<String, String> response = entry.getPlugin().process(outputs);
        // write to the database.
        System.out.println(response);
        return response;
    }
}

Can anyone tell me whether there is any problem with the above approach or is there any better and efficient way of doing the same thing? I am not sure whether there is any thread safety issue as well.

Any help will be appreciated on this.

\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

I need to call all those 5 Bundles process method in parallel using multithreaded code and then write to the database. I am not sure what is the right way to do that? Should I have five thread? One thread for each bundle? But what will happen in that scenario, suppose if I have 50 bundles, then I will have 50 threads?

It is recommended not to have more threads than the number of physical cpu's on that machine in order to be efficient. So if you have 50 Bundles and 8 cores for instances than an ExecutorService with 8 threads should be enough. The bundles will be executed 8 at a time and at most there will be 50/8 rounds of execution. (In real world some of the bundles will execute faster, etc.)

And also, I want to have timeout feature as well. If any bundles is taking lot of time than the threshold setup by us, then it should get timeout and log as an error that this bundle has taken lot of time.

You could use a Timer (ScheduledExecutor) on which you register the action to take place in case of a timeout. On timeout you should iterate the list of futures and cancel the ones that are not done yet.

Otherwise the code looks good enough. I can't see why should not be optimal.

\$\endgroup\$
8
  • \$\begingroup\$ Thanks csoroiu for the suggestion. So you are saying, I can have executorservice with fixedthreadpool size of 8 in general? Right? And also, I am using timeout feature as well. But you are suggesting to use Timer instead? If yes, can you provide an example for that basis on my above code? \$\endgroup\$
    – arsenal
    Sep 5, 2013 at 20:43
  • \$\begingroup\$ And also I am not able to understand- What I am supposed to log here? if (!future.isDone()) { // log error for this entry } meaning what kind of error, I am supposed to log here? \$\endgroup\$
    – arsenal
    Sep 5, 2013 at 20:51
  • \$\begingroup\$ I wasn't paying too much attention to the timeout parameter, but code should be good enough. My suggestion is to use a ScheduledExecutor (basically will do the same thing that timeout parameter does in case of invokeAll). The number 8 is relative to the number of cores available on your machine. So if you have a machine with 8 cores, 8 is a good number. \$\endgroup\$
    – Claudiu
    Sep 5, 2013 at 20:52
  • \$\begingroup\$ Cool. Thanks. And how about this one- What I am supposed to log here? if (!future.isDone()) { // log error for this entry } meaning what kind of error, I am supposed to log here? And also- in the catch block of InterruptedException, Am I doing the right thing? \$\endgroup\$
    – arsenal
    Sep 5, 2013 at 20:54
  • \$\begingroup\$ if (!future.isDone()) { // log error for this entry } - here you may log information related to the entry that was not done. You might log the name of the bundle(plugin) that did not finished. This information is useful for debug purposes. Additionally, for same debug purposes, it might be useful to log the execution time for each of the bundles inside ProcessBundleHolderEntry.call method. You can use System.currentTimeMillis() for this. \$\endgroup\$
    – Claudiu
    Sep 5, 2013 at 21:02

Your Answer

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

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