0
\$\begingroup\$

In my Java 8 application there are 3 different classes but each of them contains one similar method, which is dedicated to parallelization of the business logic execution. It's important to note, two of these three methods are called from another static methods, while the last one from instance-based method. Below, I provide the source code of these 3 parallelizationEngine() methods:

parallelizationEngine #1

private static void parallelizationEngine(int cpuCoresNum,
                                          ExecutorService service,
                                          MongoCollection<Document> dbC_Episodes,
                                          MongoCollection<Document> dbC_Events,
                                          ArrayList<Document> unprocessedEpisodes,
                                          ArrayList<Document> unprocessedEvents) {

    // update status of the queried items
    DBAgent.updateItemsStatus(dbC_Episodes, unprocessedEpisodes, FetchStatus.IN_PROCESS_META.getID());
    DBAgent.updateItemsStatus(dbC_Events, unprocessedEvents, FetchStatus.IN_PROCESS_META.getID());

    int itemsPerCore = unprocessedEvents.size() / cpuCoresNum;

    // prevent infinite thread generation if num of unprocessed items is less than number of CPU cores
    if (itemsPerCore == 0) {
        itemsPerCore = 1;
    }

    int itNum = 1;
    int upperIndex = 0;

    ArrayList<Callable<Boolean>> callables = new ArrayList<>();

    for (int i = 0; i < unprocessedEvents.size(); i += itemsPerCore) {

        int indexFrom = i;
        upperIndex += itemsPerCore;
        int toIndex = upperIndex;

        // all remaining items will be processed in the last thread
        if (itNum == cpuCoresNum) {
            toIndex = unprocessedEvents.size();
            i = unprocessedEvents.size();
        }

        int indexTo = toIndex;
        int treadID = itNum++;

        callables.add(() -> {
            Thread.currentThread().setName("im_metaDataFiller_#" + String.format("%03d", treadID));
            metaDataEngine(dbC_Episodes, dbC_Events, new ArrayList<>(unprocessedEpisodes.subList(indexFrom, indexTo)), unprocessedEvents);
            return true;
        });
    }

    try {
        // check if every thread has been processed correctly
        List<Future<Boolean>> futures = service.invokeAll(callables);

        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (ExecutionException e) {
                Logger.printMsg(true, "Exception", "dba.MetaDataFiller.parallelizationEngine(): ", e.getClass().getName() + " : " + e.getMessage());
            }
        }
    } catch (InterruptedException e) {
        Logger.printMsg(true, "Exception", "dba.MetaDataFiller.parallelizationEngine(): ", e.getClass().getName() + " : " + e.getMessage());
    }
}

parallelizationEngine #2

private static void parallelizationEngine(int cpuCoresNum,
                                          ExecutorService service,
                                          MongoCollection<Document> dbC_Events,
                                          ArrayList<Document> unprocessedEvents) {

    // update status of the queried items
    DBAgent.updateItemsStatus(dbC_Events, unprocessedEvents, FetchStatus.IN_PROCESS_NLP.getID());

    StanfordCoreNLP pipeline = loadNLPPipeline();
    NERClassifierCombiner nerCombClassifier = loadNERClassifiers(getNERClassPath());

    int itemsPerCore = unprocessedEvents.size() / cpuCoresNum;

    // prevent infinite thread generation if num of unprocessed items is less than number of CPU cores
    if (itemsPerCore == 0) {
        itemsPerCore = 1;
    }

    int itNum = 1;
    int upperIndex = 0;

    ArrayList<Callable<Boolean>> callables = new ArrayList<>();

    for (int i = 0; i < unprocessedEvents.size(); i += itemsPerCore) {

        int indexFrom = i;
        upperIndex += itemsPerCore;
        int toIndex = upperIndex;

        // all remaining items will be processed in the last thread
        if (itNum == cpuCoresNum) {
            toIndex = unprocessedEvents.size();
            i = unprocessedEvents.size();
        }

        int indexTo = toIndex;
        int treadID = itNum++;

        callables.add(() -> {
            Thread.currentThread().setName("im_nlpAnalyzer_#" + String.format("%03d", treadID));
            analyzerEngine(pipeline, nerCombClassifier, dbC_Events, new ArrayList<>(unprocessedEvents.subList(indexFrom, indexTo)));
            return true;
        });
    }

    try {
        // check if every thread has been processed correctly
        List<Future<Boolean>> futures = service.invokeAll(callables);

        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (ExecutionException e) {
                Logger.printMsg(true, "Exception", "ir.NLPAnalyzer.parallelizationEngine(): ", e.getClass().getName() + " : " + e.getMessage());
            }
        }
    } catch (InterruptedException e) {
        Logger.printMsg(true, "Exception", "ir.NLPAnalyzer.parallelizationEngine(): ", e.getClass().getName() + " : " + e.getMessage());
    }
}

parallelizationEngine #3

private void parallelizationEngine(int cpuCoresNum,
                                   ExecutorService service,
                                   MongoCollection<Document> dbC_Events,
                                   ArrayList<Document> unprocessedEvents) {

    // update status of queried items
    DBAgent.updateItemsStatus(dbC_Events, unprocessedEvents, FetchStatus.IN_PROCESS_FETCH.getID());

    int itemsPerCore = unprocessedEvents.size() / cpuCoresNum;

    // prevent infinite thread generation if num of unprocessed items is less than number of CPU cores
    if (itemsPerCore == 0) {
        itemsPerCore = 1;
    }

    int itNum = 1;
    int upperIndex = 0;

    ArrayList<Callable<Boolean>> callables = new ArrayList<>();

    for (int i = 0; i < unprocessedEvents.size(); i += itemsPerCore) {

        int indexFrom = i;
        upperIndex += itemsPerCore;
        int toIndex = upperIndex;

        // all remaining items will be processed in the last thread
        if (itNum == cpuCoresNum) {
            toIndex = unprocessedEvents.size();
            i = unprocessedEvents.size();
        }

        int indexTo = toIndex;
        int treadID = itNum++;

        callables.add(() -> {
            Thread.currentThread().setName("im_evFullTextImg_#" + String.format("%03d", treadID));
            this.fetcherEngine(dbC_Events, new ArrayList<>(unprocessedEvents.subList(indexFrom, indexTo)));
            return true;
        });
    }

    try {
        // check if every thread has been processed correctly
        List<Future<Boolean>> futures = service.invokeAll(callables);

        for (Future<Boolean> future : futures) {
            try {
                future.get();
            } catch (ExecutionException e) {
                Logger.printMsg(true, "Exception", "crawler.FullTextImgFetcher.parallelizationEngine(): ", e.getClass().getName() + " : " + e.getMessage());
            }
        }
    } catch (InterruptedException e) {
        Logger.printMsg(true, "Exception", "crawler.FullTextImgFetcher.parallelizationEngine(): ", e.getClass().getName() + " : " + e.getMessage());
    }
}

As you may see, these three methods are quite similar, and I think it will be correct to refactor this code in order to eliminate extra code duplication.

The problem is that two of three methods are static, and one is instance-based. Bearing in mind this fact, is there any way to extract the duplicated code to the separate entity and to pass the method I want to execute within callables.add() as a parameter?

\$\endgroup\$
  • 1
    \$\begingroup\$ Make life easier on yourself and stop using static for anything at all. It should be banned from the language. \$\endgroup\$ – Software Engineer Apr 27 '16 at 14:48
1
\$\begingroup\$

You can move your method to a Utils-class and pass instances of different Classes all implementing the same Interface. e.g.

Based on your parallelizationEngine #3:

Utils:

public static void parallelizationEngine(int cpuCoresNum,
                               ExecutorService service,
                               MongoCollection<Document> dbC_Events,
                               ArrayList<Document> unprocessedEvents,
                               AddHandler addHandler) {

  // update status of queried items
  DBAgent.updateItemsStatus(dbC_Events, unprocessedEvents, FetchStatus.IN_PROCESS_FETCH.getID());

  int itemsPerCore = unprocessedEvents.size() / cpuCoresNum;

  // prevent infinite thread generation if num of unprocessed items is less than number of CPU cores
  if (itemsPerCore == 0) {
    itemsPerCore = 1;
  }

  int itNum = 1;
  int upperIndex = 0;

  ArrayList<Callable<Boolean>> callables = new ArrayList<>();
  for (int i = 0; i < unprocessedEvents.size(); i += itemsPerCore) {

    int indexFrom = i;
    upperIndex += itemsPerCore;
    int toIndex = upperIndex;

    // all remaining items will be processed in the last thread
    if (itNum == cpuCoresNum) {
        toIndex = unprocessedEvents.size();
        i = unprocessedEvents.size();
    }

    int indexTo = toIndex;
    int treadID = itNum++;

    callables.add(() -> {
        Thread.currentThread().setName("im_metaDataFiller_#" + String.format("%03d", treadID));
        addHandler.doSomething();  // HERE DEPENDING ON THE OPPARAMETER 'addHandler', DIFFERENT THINGS HAPPEN
        return true;
    });
  }

  try {
    // check if every thread has been processed correctly
    List<Future<Boolean>> futures = service.invokeAll(callables);

    for (Future<Boolean> future : futures) {
        try {
            future.get();
        } catch (ExecutionException e) {
            Logger.printMsg(true, "Exception", "dba.MetaDataFiller.performMetaDataComputing(): ", e.getClass().getName() + " : " + e.getMessage());
        }
    }
  } catch (InterruptedException e) {
    Logger.printMsg(true, "Exception", "dba.MetaDataFiller.performMetaDataComputing(): ", e.getClass().getName() + " : " + e.getMessage());
  }
}


interface AddHandler{
  public void doSometing();
}

And classes implementing the Interface:

class A implements AddHandler{

  public void doSometing(){
    .. so something 
  }
}

class B implements AddHandler{

  public void doSometing(){
    .. so something different form class A
  }
}

Note you need to do more!

  • Because the number of parameters of your different parallelizationEngine(..) differ a bit you can write a universal method with all parameters needed by any implementation and than writhe warp this in methods with only the required parameters.

  • Because at the begin of your different parallelizationEngine(..) methods also differs a bit you can also extract this part to method of a parameter (like I showed for the part in the for-loop) to make all your implementations exactly the same so you can reduce them to only one implementation.

\$\endgroup\$
  • \$\begingroup\$ Do I understand correctly, that I replace %DIFFERENT_METHOD_TO_CALL% with addHandler.doSomething() and call within the class A the Utils.processEvents(unprocessedEvents, this)? \$\endgroup\$ – Mike B. Apr 25 '16 at 12:41
  • 1
    \$\begingroup\$ @Mike B.: replace %DIFFERENT_METHOD_TO_CALL% with addHandler.doSomething() and tin all classes where you now have the 'similar' code, simply call Utils.processEvents(unprocessedEvents, this) . You did not add the method-call in your code, so I assumed that it is a method with a name like 'processEvents' and since it loops over a 'unprocessedEvents' of I assumed that these might be a parameter of the method. \$\endgroup\$ – MrSmith42 Apr 25 '16 at 13:37
  • \$\begingroup\$ I recomposed the question with richer code examples. Please, have a look at new version. \$\endgroup\$ – Mike B. Apr 25 '16 at 16:01
  • \$\begingroup\$ @Mike B.: I updated the code in my answer according to your updated question and also added 2 points about the different parameter-list and the different code at the begin of the methods. \$\endgroup\$ – MrSmith42 Apr 27 '16 at 10:16
  • \$\begingroup\$ Thanks for the update. One more moment, how can I pass the AddHandler addHandler from the static method to Utils.parallelizationEngine? If it was not static method I would pass this but here there is a static method. \$\endgroup\$ – Mike B. Apr 27 '16 at 11:00

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.