package fr.liglab.lcm.io;

import fr.liglab.lcm.internals.Dataset;
import fr.liglab.lcm.internals.FrequentsIterator;

public class PerItemTopKCollectorThreadSafeInitialized extends
		PerItemGroupTopKCollector {
	/*
	 * If we go for threads, how do we make this threadsafe ? If this is not a
	 * bottleneck, synchronized on collect is ok Else, if we want parallelism,
	 * we need to take a lock when dealing with an item We can use a collection
	 * of locks and a hash of the item to do that safely, be careful with
	 * insertions in the map though
	 * 
	 * What we should do for MapReduce is create it in the open and close it on
	 * the close, so when we execute several maps on the same mapper they all
	 * benefit from the top-k of the others
	 */
	public PerItemTopKCollectorThreadSafeInitialized(
			final PatternsCollector decorated, Dataset dataset, final int k) {
		this(decorated, k, dataset, false);
	}
	
	public PerItemTopKCollectorThreadSafeInitialized(
			final PatternsCollector follower, final int k, Dataset dataset,
			final boolean outputEachPatternOnce) {
		
		this(follower, k, dataset.counters.getFrequentsIterator(), outputEachPatternOnce);
	}
	
	public PerItemTopKCollectorThreadSafeInitialized(
			final PatternsCollector follower, final int k, FrequentsIterator items,
			final boolean outputEachPatternOnce) {
		super(follower, k);
		// we may want to hint a default size, it is at least the group size,
		// but in practice much bigger
		this.init(items);
	}
	
	public PerItemTopKCollectorThreadSafeInitialized(
			final PatternsCollector follower, final int k, Dataset dataset,
			final boolean mineInGroup, final boolean mineOutGroup) {
		
		super(follower, k, mineInGroup, mineOutGroup);
		
		// FIXME - maybe this would better happen in setGroup ?
		this.init(dataset.counters.getFrequentsIterator());
	}
	
	private void init(final FrequentsIterator iterator) {
		for (int item = iterator.next(); item != -1; item = iterator.next()) {
			this.topK.put(item, new PatternWithFreq[k]);
		}
	}
	
	@Override
	protected void insertPatternInTop(final int support, final int[] pattern,
			final int item) {
		PatternWithFreq[] itemTopK = this.topK.get(item);
		if (itemTopK == null) {
			throw new RuntimeException("item not initialized " + item);
		} else {
			synchronized (itemTopK) {
				super.updateTop(support, pattern, itemTopK);
			}
		}
	}
}

@@ -173,19 +173,17 @@
  extends
-%0A%09%09
+ 
 PerItemG
@@ -784,36 +784,32 @@
 SafeInitialized(
-%0A%09%09%09
 final PatternsCo
@@ -896,25 +896,24 @@
  false);%0A%09%7D%0A
-%09
 %0A%09public Per
@@ -943,36 +943,32 @@
 SafeInitialized(
-%0A%09%09%09
 final PatternsCo
@@ -1057,18 +1057,16 @@
 Once) %7B%0A
-%09%09
 %0A%09%09this(
@@ -1142,25 +1142,24 @@
 rnOnce);%0A%09%7D%0A
-%09
 %0A%09public Per
@@ -1189,36 +1189,32 @@
 SafeInitialized(
-%0A%09%09%09
 final PatternsCo
@@ -1243,17 +1243,20 @@
 l int k,
- 
+%0A%09%09%09
 Frequent
@@ -1271,20 +1271,17 @@
 r items,
-%0A%09%09%09
+ 
 final bo
@@ -1461,17 +1461,16 @@
 ms);%0A%09%7D%0A
-%09
 %0A%09public
@@ -1512,20 +1512,16 @@
 ialized(
-%0A%09%09%09
 final Pa
@@ -1636,18 +1636,16 @@
 roup) %7B%0A
-%09%09
 %0A%09%09super
@@ -1686,18 +1686,16 @@
 Group);%0A
-%09%09
 %0A%09%09// FI
@@ -1802,17 +1802,16 @@
 ());%0A%09%7D%0A
-%09
 %0A%09privat
@@ -1986,17 +1986,16 @@
 %0A%09%09%7D%0A%09%7D%0A
-%09
 %0A%09@Overr
@@ -2072,20 +2072,17 @@
 pattern,
-%0A%09%09%09
+ 
 final in
@@ -2286,14 +2286,23 @@
 per.
-update
+insertPatternIn
 Top(
@@ -2323,20 +2323,16 @@
 rn, item
-TopK
 );%0A%09%09%09%7D%0A

fix heritage