Package it.unimi.di.law.bubing.sieve

Class AbstractSieve<K,V>

java.lang.Object

it.unimi.di.law.bubing.sieve.AbstractSieve<K,V>

All Implemented Interfaces:

java.io.Closeable, java.lang.AutoCloseable

Direct Known Subclasses:

IdentitySieve, MercatorSieve

public abstract class AbstractSieve<K,V>
extends java.lang.Object
implements java.io.Closeable

A sort of a map, that handles (key,value) pairs of generic type. You can think of it as a long-term sieve where you can put the same k as many times as you want, but that provides the following guarantees:

• every key that is enqueued will eventually be dequeued;
• only enqueued keys will be dequeued;
• every key will be dequeued just once.

Broadly, you can think of all the sieve logic as if it was like a uniq filter (in the sense of UN*X). Here we are completely disregarding the asynchronous logic of dequeuing, which of course impacts the behaviour.

This data structure generalizes and adapts that described in the paper IRLbot: Scaling to 6 Billion Pages and Beyond by Lee et al., where it was called DRUM (Disk Repository with Update Management). We prefer not to refer anymore to disk in our abstract version because different implementations may decide not to use the disk in a strict sense to hold the items.

The structure was originally described as follows:

• it is basically a way to store (k,v) pairs on an external data structure (we prefer to call them k and v instead of keys and values because we shall use the word key with a different meaning); k's present in the store are unique (that is, no two pairs in the store can have the same k);
• the data structure offers three kinds of operations, called check, update and check+update; typically, such operations are batched (that is, they are performed on a large input), but for the moment we prefer to describe them as if they were performed on a single input:
  • check(k,v,x) takes a triple (k,v,x) (x is called the auxiliary data), and determines if k is new (i.e., no pair with k is present in the store already) or not; if it is new, the check operation just returns the triple itself (k,v,x); otherwise, it returns (k,v,x,w) where w is the value that is associated with k in the store;
  • update(k,v,x) if k is new, adds the pair (k,v) to the store; otherwise, i.e. if some pair (k,w) is present in the store, it changes it to (k,up(k,v,x,w)) where up is some suitable update function;
  • check+update(k,v,x) performs a check followed by an update, and returns what the check part would return;
• actually, as we said, all operations are performed in batch and asynchronously; so, for example, check typically takes as input a batch of triples, and divides it into two flows: one is a copy of the triples containing new k's, the other is made of quadruples for duplicated k's (note that if some k appears more than once in the input batch, only the first time it may be considered as new). These two flows are asynchronously accessible through some other method (see below).

This implementation of the data structure is different from the one described above in the following regards:

• we assume (as implicitly done in the original paper) that k is a suitable hash of x, so all functions only take as input (x,v) (where x is called key and v is called a value) and k is produced as some hash applied to x (the hashing strategy is provided at construction time);
• only the check+update operation is available, with the following additional limitation: the flow of quadruples (i.e., the flow containing duplicate k's) is simply discarded; the update strategy (the function up described above) is by default defined by up(k,v,x,w)=v (i.e., substitute the old value with the new one), but may be changed (specifying a different update strategy at construction time): observe that, actually, up doesn't take (k,v,x,w) as input but rather (x,v,w), because of the above described change in the treatment of keys/auxiliary data;
• the call to check+update is actually decoupled into two phases: the method enqueue(Object, Object) corresponds to the call: it should be very fast, although some implementation may require it to be blocking, and will perform the check (and update) sometime in the future; when a flush() is performed, a new flow of pairs (the pairs that the check method would return) is created: those pairs are actually provided to a AbstractSieve.NewFlowReceiver object that can decide what to do with them (see the AbstractSieve.NewFlowReceiver documentation).

Thread safety

Implementors should guarantee that all method calls are thread-safe, and as efficient as possible. The enqueue(Object, Object) method should usually be non-blocking, but some bounded implementations may decide to make it wait until at least one element is dequeued: this event should be carefully documented. A call to flush() guarantees that all new elements that have not yet been dequeued are made available, if any.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
static class	AbstractSieve.DefaultUpdateStrategy<K,V>
static class	AbstractSieve.DiskNewFlow<T>	A basic, on-disk AbstractSieve.NewFlowReceiver.
static interface	AbstractSieve.NewFlowReceiver<K>	An object that can receive a new flow of hash/key pairs and that acts as a listener for the AbstractSieve.
static class	AbstractSieve.SieveEntry<K,V>	A (key,value) pair.
static interface	AbstractSieve.UpdateStrategy<K,V>	An update strategy: it determines how a stored value should be updated in the presence of duplicate keys.

Field Summary

Fields

Modifier and Type	Field	Description
static it.unimi.dsi.sux4j.mph.AbstractHashFunction<java.lang.CharSequence>	CHAR_SEQUENCE_HASHING_STRATEGY
protected it.unimi.dsi.sux4j.mph.AbstractHashFunction<K>	hashingStrategy
protected ByteSerializerDeserializer<K>	keySerDeser
protected AbstractSieve.NewFlowReceiver<K>	newFlowReceiver
protected AbstractSieve.UpdateStrategy<K,V>	updateStrategy
protected ByteSerializerDeserializer<V>	valueSerDeser

Constructor Summary

Constructors

Constructor	Description
AbstractSieve(ByteSerializerDeserializer<K> keySerDeser, ByteSerializerDeserializer<V> valueSerDeser, it.unimi.dsi.sux4j.mph.AbstractHashFunction<K> hashingStrategy, AbstractSieve.UpdateStrategy<K,V> updateStrategy)	Creates a new sieve with the given data.

Method Summary

Modifier and Type	Method	Description
abstract void	close()	Closes (forever) this sieve.
abstract boolean	enqueue(K key, V value)	Add the given (key,value) pair to the store.
abstract void	flush()	Forces the check+update of all pairs that have been enqueued.
void	setNewFlowRecevier(AbstractSieve.NewFlowReceiver<K> newFlowReceiver)	Sets the receiver for the new flows generated by this sieve.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

keySerDeser

protected final ByteSerializerDeserializer<K> keySerDeser

valueSerDeser

protected final ByteSerializerDeserializer<V> valueSerDeser

hashingStrategy

protected final it.unimi.dsi.sux4j.mph.AbstractHashFunction<K> hashingStrategy

updateStrategy

protected final AbstractSieve.UpdateStrategy<K,V> updateStrategy

newFlowReceiver

protected AbstractSieve.NewFlowReceiver<K> newFlowReceiver

CHAR_SEQUENCE_HASHING_STRATEGY

public static final it.unimi.dsi.sux4j.mph.AbstractHashFunction<java.lang.CharSequence> CHAR_SEQUENCE_HASHING_STRATEGY

Constructor Detail

AbstractSieve

public AbstractSieve(ByteSerializerDeserializer<K> keySerDeser,
                     ByteSerializerDeserializer<V> valueSerDeser,
                     it.unimi.dsi.sux4j.mph.AbstractHashFunction<K> hashingStrategy,
                     AbstractSieve.UpdateStrategy<K,V> updateStrategy)

Creates a new sieve with the given data.

Parameters:

keySerDeser - the serializer and deserializer to be used to store keys.

valueSerDeser - the serializer and deserializer to be used to store values.

hashingStrategy - the function to be applied to keys (CharSequence) to obtain hash values (the store actually contains hash values, not keys).

updateStrategy - the strategy used to update the values associated to duplicate keys.

Method Detail

enqueue

public abstract boolean enqueue(K key,
                                V value)
                         throws java.io.IOException,
                                java.lang.InterruptedException

Add the given (key,value) pair to the store.

Returns:

true if the operation caused a flush.

Throws:

java.io.IOException

java.lang.InterruptedException

close

public abstract void close()
                    throws java.io.IOException

Closes (forever) this sieve.

Specified by:

close in interface java.lang.AutoCloseable

Specified by:

close in interface java.io.Closeable

Throws:

java.io.IOException

setNewFlowRecevier

public void setNewFlowRecevier(AbstractSieve.NewFlowReceiver<K> newFlowReceiver)

Sets the receiver for the new flows generated by this sieve.

Parameters:

newFlowReceiver - the new flow receiver for this sieve.

flush

public abstract void flush()
                    throws java.io.IOException,
                           java.lang.InterruptedException

Forces the check+update of all pairs that have been enqueued.

Throws:

java.io.IOException

java.lang.InterruptedException