MultithreadedBatchItemSimilarities extends BatchItemSimilarities { @Override public int computeItemSimilarities(int degreeOfParallelism, int maxDurationInHours, SimilarItemsWriter writer) throws IOException { ExecutorService executorService = Executors.newFixedThreadPool(degreeOfParallelism + 1); Output output = null; try { writer.open(); DataModel dataModel = getRecommender().getDataModel(); BlockingQueue<long[]> itemsIDsInBatches = queueItemIDsInBatches(dataModel, batchSize, degreeOfParallelism); BlockingQueue<List<SimilarItems>> results = new LinkedBlockingQueue<>(); AtomicInteger numActiveWorkers = new AtomicInteger(degreeOfParallelism); for (int n = 0; n < degreeOfParallelism; n++) { executorService.execute(new SimilarItemsWorker(n, itemsIDsInBatches, results, numActiveWorkers)); } output = new Output(results, writer, numActiveWorkers); executorService.execute(output); } catch (Exception e) { throw new IOException(e); } finally { executorService.shutdown(); try { boolean succeeded = executorService.awaitTermination(maxDurationInHours, TimeUnit.HOURS); if (!succeeded) { throw new RuntimeException("Unable to complete the computation in " + maxDurationInHours + " hours!"); } } catch (InterruptedException e) { throw new RuntimeException(e); } Closeables.close(writer, false); } return output.getNumSimilaritiesProcessed(); } MultithreadedBatchItemSimilarities(ItemBasedRecommender recommender, int similarItemsPerItem);  MultithreadedBatchItemSimilarities(ItemBasedRecommender recommender, int similarItemsPerItem, int batchSize); @Override int computeItemSimilarities(int degreeOfParallelism, int maxDurationInHours, SimilarItemsWriter writer);  }
@Test public void lessItemsThanBatchSize() throws Exception { FastByIDMap<PreferenceArray> userData = new FastByIDMap<>(); userData.put(1, new GenericUserPreferenceArray(Arrays.asList(new GenericPreference(1, 1, 1), new GenericPreference(1, 2, 1), new GenericPreference(1, 3, 1)))); userData.put(2, new GenericUserPreferenceArray(Arrays.asList(new GenericPreference(2, 1, 1), new GenericPreference(2, 2, 1), new GenericPreference(2, 4, 1)))); DataModel dataModel = new GenericDataModel(userData); ItemBasedRecommender recommender = new GenericItemBasedRecommender(dataModel, new TanimotoCoefficientSimilarity(dataModel)); BatchItemSimilarities batchSimilarities = new MultithreadedBatchItemSimilarities(recommender, 10); batchSimilarities.computeItemSimilarities(1, 1, mock(SimilarItemsWriter.class)); }
@Test(expected = IOException.class) public void higherDegreeOfParallelismThanBatches() throws Exception { FastByIDMap<PreferenceArray> userData = new FastByIDMap<>(); userData.put(1, new GenericUserPreferenceArray(Arrays.asList(new GenericPreference(1, 1, 1), new GenericPreference(1, 2, 1), new GenericPreference(1, 3, 1)))); userData.put(2, new GenericUserPreferenceArray(Arrays.asList(new GenericPreference(2, 1, 1), new GenericPreference(2, 2, 1), new GenericPreference(2, 4, 1)))); DataModel dataModel = new GenericDataModel(userData); ItemBasedRecommender recommender = new GenericItemBasedRecommender(dataModel, new TanimotoCoefficientSimilarity(dataModel)); BatchItemSimilarities batchSimilarities = new MultithreadedBatchItemSimilarities(recommender, 10); batchSimilarities.computeItemSimilarities(2, 1, mock(SimilarItemsWriter.class)); fail(); }
@Test public void testCorrectNumberOfOutputSimilarities() throws Exception { FastByIDMap<PreferenceArray> userData = new FastByIDMap<>(); userData.put(1, new GenericUserPreferenceArray(Arrays.asList(new GenericPreference(1, 1, 1), new GenericPreference(1, 2, 1), new GenericPreference(1, 3, 1)))); userData.put(2, new GenericUserPreferenceArray(Arrays.asList(new GenericPreference(2, 1, 1), new GenericPreference(2, 2, 1), new GenericPreference(2, 4, 1)))); DataModel dataModel = new GenericDataModel(userData); ItemBasedRecommender recommender = new GenericItemBasedRecommender(dataModel, new TanimotoCoefficientSimilarity(dataModel)); BatchItemSimilarities batchSimilarities = new MultithreadedBatchItemSimilarities(recommender, 10, 2); int numOutputSimilarities = batchSimilarities.computeItemSimilarities(2, 1, mock(SimilarItemsWriter.class)); assertEquals(numOutputSimilarities, 10); }