You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by pm...@apache.org on 2009/03/02 08:57:31 UTC
svn commit: r749218 [33/34] - in /incubator/cassandra: branches/ dist/
nightly/ site/ tags/ trunk/ trunk/lib/ trunk/src/ trunk/src/org/
trunk/src/org/apache/ trunk/src/org/apache/cassandra/
trunk/src/org/apache/cassandra/analytics/ trunk/src/org/apache...
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/BloomFilter.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/BloomFilter.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/BloomFilter.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/BloomFilter.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,633 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils;
+
+import java.math.*;
+import java.nio.ByteBuffer;
+import java.nio.LongBuffer;
+import java.io.*;
+import java.security.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.zip.*;
+
+import javax.xml.bind.annotation.XmlElement;
+
+import org.apache.cassandra.io.DataInputBuffer;
+import org.apache.cassandra.io.DataOutputBuffer;
+import org.apache.cassandra.io.ICompactSerializer;
+import org.apache.cassandra.io.SSTable;
+
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class BloomFilter implements Serializable
+{
+ public static class CountingBloomFilter implements Serializable
+ {
+ private static ICompactSerializer<CountingBloomFilter> serializer_;
+ static
+ {
+ serializer_ = new CountingBloomFilterSerializer();
+ }
+
+ public static ICompactSerializer<CountingBloomFilter> serializer()
+ {
+ return serializer_;
+ }
+
+ @XmlElement(name="Filter")
+ private byte[] filter_ = new byte[0];
+
+ @XmlElement(name="Size")
+ private int size_;
+
+ @XmlElement(name="Hashes")
+ private int hashes_;
+
+ /* Keeps count of number of keys added to CBF */
+ private transient int count_ = 0;
+ private transient Random random_ = new Random(System.currentTimeMillis());
+
+ /*
+ * This is just for JAXB.
+ */
+ private CountingBloomFilter()
+ {
+ }
+
+ public CountingBloomFilter(int numElements, int bitsPerElement)
+ {
+ // TODO -- think about the trivial cases more.
+ // Note that it should indeed be possible to send a bloom filter that
+ // encodes the empty set.
+ if (numElements < 0 || bitsPerElement < 1)
+ throw new IllegalArgumentException("Number of elements and bits "
+ + "must be non-negative.");
+ // Adding a small random number of bits so that even if the set
+ // of elements hasn't changed, we'll get different false positives.
+ size_ = numElements * bitsPerElement + 20 + random_.nextInt(64);
+ filter_ = new byte[size_];
+ hashes_ = BloomCalculations.computeBestK(bitsPerElement);
+ }
+
+ CountingBloomFilter(int size, int hashes, byte[] filter)
+ {
+ size_ = size;
+ hashes_ = hashes;
+ filter_ = filter;
+ }
+
+ public CountingBloomFilter cloneMe()
+ {
+ byte[] filter = new byte[filter_.length];
+ System.arraycopy(filter_, 0, filter, 0, filter_.length);
+ return new BloomFilter.CountingBloomFilter(size_, hashes_, filter);
+ }
+
+ int size()
+ {
+ return size_;
+ }
+
+ int hashes()
+ {
+ return hashes_;
+ }
+
+ byte[] filter()
+ {
+ return filter_;
+ }
+
+ public BloomFilter.CountingBloomFilter merge(BloomFilter.CountingBloomFilter cbf)
+ {
+ if ( cbf == null )
+ return this;
+
+ if ( size_ >= cbf.size_ )
+ {
+ for ( int i = 0; i < cbf.filter_.length; ++i )
+ {
+ filter_[i] |= cbf.filter_[i];
+ }
+ return this;
+ }
+ else
+ {
+ for ( int i = 0; i < filter_.length; ++i )
+ {
+ cbf.filter_[i] |= filter_[i];
+ }
+ return cbf;
+ }
+ }
+
+ public boolean isPresent(String key)
+ {
+ boolean bVal = true;
+ for (int i = 0; i < hashes_; ++i)
+ {
+ ISimpleHash hash = hashLibrary_.get(i);
+ int hashValue = hash.hash(key);
+ int index = Math.abs(hashValue % size_);
+ if (filter_[index] == 0)
+ {
+ bVal = false;
+ break;
+ }
+ }
+ return bVal;
+ }
+
+ /*
+ param@ key -- value whose hash is used to fill
+ the filter_.
+ This is a general purpose API.
+ */
+ public void add(String key)
+ {
+ if ( !isPresent(key) )
+ ++count_;
+ for (int i = 0; i < hashes_; ++i)
+ {
+ ISimpleHash hash = hashLibrary_.get(i);
+ int hashValue = hash.hash(key);
+ int index = Math.abs(hashValue % size_);
+ byte value = (filter_[index] == 0xFF) ? filter_[index] : (byte)( (++filter_[index]) & 0xFF );
+ filter_[index] = value;
+ }
+ }
+
+ public boolean delete(String key)
+ {
+ boolean bVal = isPresent(key);
+ if ( !bVal )
+ {
+ --count_;
+ return bVal;
+ }
+
+ for (int i = 0; i < hashes_; ++i)
+ {
+ ISimpleHash hash = hashLibrary_.get(i);
+ int hashValue = hash.hash(key);
+ int index = Math.abs(hashValue % size_);
+ byte value = (filter_[index] == 0) ? filter_[index] : (byte)( (--filter_[index]) & 0xFF );
+ filter_[index] = value;
+ }
+
+ return bVal;
+ }
+
+ public int count()
+ {
+ return count_;
+ }
+ }
+
+ private static List<ISimpleHash> hashLibrary_ = new ArrayList<ISimpleHash>();
+ private static ICompactSerializer<BloomFilter> serializer_;
+
+ static
+ {
+ serializer_ = new BloomFilterSerializer();
+ hashLibrary_.add(new RSHash());
+ hashLibrary_.add(new JSHash());
+ hashLibrary_.add(new PJWHash());
+ hashLibrary_.add(new ELFHash());
+ hashLibrary_.add(new BKDRHash());
+ hashLibrary_.add(new SDBMHash());
+ hashLibrary_.add(new DJBHash());
+ hashLibrary_.add(new DEKHash());
+ hashLibrary_.add(new BPHash());
+ hashLibrary_.add(new FNVHash());
+ hashLibrary_.add(new APHash());
+ }
+
+ public static ICompactSerializer<BloomFilter> serializer()
+ {
+ return serializer_;
+ }
+
+ private BitSet filter_;
+ private int count_;
+ private int size_;
+ private int hashes_;
+ private Random random_ = new Random(System.currentTimeMillis());
+
+ public BloomFilter(int bitsPerElement)
+ {
+ if (bitsPerElement < 1)
+ throw new IllegalArgumentException("Number of bitsPerElement "
+ + "must be non-negative.");
+ // Adding a small random number of bits so that even if the set
+ // of elements hasn't changed, we'll get different false positives.
+ size_ = 20 + random_.nextInt(64);
+ filter_ = new BitSet(size_);
+ hashes_ = BloomCalculations.computeBestK(bitsPerElement);
+ }
+
+ public BloomFilter(int numElements, int bitsPerElement)
+ {
+ // TODO -- think about the trivial cases more.
+ // Note that it should indeed be possible to send a bloom filter that
+ // encodes the empty set.
+ if (numElements < 0 || bitsPerElement < 1)
+ throw new IllegalArgumentException("Number of elements and bits "
+ + "must be non-negative.");
+ // Adding a small random number of bits so that even if the set
+ // of elements hasn't changed, we'll get different false positives.
+ count_ = numElements;
+ size_ = numElements * bitsPerElement + 20 + random_.nextInt(64);
+ filter_ = new BitSet(size_);
+ //hashes_ = BloomCalculations.computeBestK(bitsPerElement);
+ hashes_ = 8;
+ }
+
+ public BloomFilter(int numElements, double maxFalsePosProbability)
+ {
+ if (numElements < 0)
+ throw new IllegalArgumentException("Number of elements must be "
+ + "non-negative.");
+ BloomCalculations.BloomSpecification spec = BloomCalculations
+ .computeBitsAndK(maxFalsePosProbability);
+ // Add a small random number of bits so that even if the set
+ // of elements hasn't changed, we'll get different false positives.
+ count_ = numElements;
+ size_ = numElements * spec.bitsPerElement + 20 + random_.nextInt(64);
+ filter_ = new BitSet(size_);
+ hashes_ = spec.K;
+ }
+
+ /*
+ * This version is only used by the deserializer.
+ */
+ BloomFilter(int count, int hashes, int size, BitSet filter)
+ {
+ count_ = count;
+ hashes_ = hashes;
+ size_ = size;
+ filter_ = filter;
+ }
+
+ int count()
+ {
+ return count_;
+ }
+
+ int size()
+ {
+ return size_;
+ }
+
+ int hashes()
+ {
+ return hashes_;
+ }
+
+ BitSet filter()
+ {
+ return filter_;
+ }
+
+ public BloomFilter merge(BloomFilter bf)
+ {
+ BloomFilter mergedBf = null;
+ if ( filter_.size() >= bf.filter_.size() )
+ {
+ filter_.or(bf.filter_);
+ mergedBf = this;
+ }
+ else
+ {
+ bf.filter_.or(filter_);
+ mergedBf = bf;
+ }
+ return mergedBf;
+ }
+
+ public boolean isPresent(String key)
+ {
+ boolean bVal = true;
+ for (int i = 0; i < hashes_; ++i)
+ {
+ ISimpleHash hash = hashLibrary_.get(i);
+ int hashValue = hash.hash(key);
+ int index = Math.abs(hashValue % size_);
+ if (!filter_.get(index))
+ {
+ bVal = false;
+ break;
+ }
+ }
+ return bVal;
+ }
+
+ /*
+ param@ key -- value whose hash is used to fill
+ the filter_.
+ This is a general purpose API.
+ */
+ public void fill(String key)
+ {
+ for (int i = 0; i < hashes_; ++i)
+ {
+ ISimpleHash hash = hashLibrary_.get(i);
+ int hashValue = hash.hash(key);
+ int index = Math.abs(hashValue % size_);
+ filter_.set(index);
+ }
+ }
+
+ public String toString()
+ {
+ return filter_.toString();
+ }
+
+ public static void main(String[] args) throws Throwable
+ {
+ BloomFilter bf = new BloomFilter(64*1024*1024, 15);
+ for ( int i = 0; i < 64*1024*1024; ++i )
+ {
+ bf.fill(Integer.toString(i));
+ }
+ System.out.println("Done filling ...");
+ for ( int i = 0; i < 64*1024*1024; ++i )
+ {
+ if ( !bf.isPresent(Integer.toString(i)) )
+ System.out.println("Oops");
+ }
+ }
+}
+
+class BloomFilterSerializer implements ICompactSerializer<BloomFilter>
+{
+ /*
+ * The following methods are used for compact representation
+ * of BloomFilter. This is essential, since we want to determine
+ * the size of the serialized Bloom Filter blob before it is
+ * populated armed with the knowledge of how many elements are
+ * going to reside in it.
+ */
+
+ public void serialize(BloomFilter bf, DataOutputStream dos) throws IOException
+ {
+ /* write out the count of the BloomFilter */
+ dos.writeInt(bf.count());
+ /* write the number of hash functions used */
+ dos.writeInt(bf.hashes());
+ /* write the size of the BloomFilter */
+ dos.writeInt(bf.size());
+ BitSet.serializer().serialize(bf.filter(), dos);
+ }
+
+ public BloomFilter deserialize(DataInputStream dis) throws IOException
+ {
+ /* read the count of the BloomFilter */
+ int count = dis.readInt();
+ /* read the number of hash functions */
+ int hashes = dis.readInt();
+ /* read the size of the bloom filter */
+ int size = dis.readInt();
+ BitSet bs = BitSet.serializer().deserialize(dis);
+ return new BloomFilter(count, hashes, size, bs);
+ }
+}
+
+class CountingBloomFilterSerializer implements ICompactSerializer<BloomFilter.CountingBloomFilter>
+{
+ /*
+ * The following methods are used for compact representation
+ * of BloomFilter. This is essential, since we want to determine
+ * the size of the serialized Bloom Filter blob before it is
+ * populated armed with the knowledge of how many elements are
+ * going to reside in it.
+ */
+
+ public void serialize(BloomFilter.CountingBloomFilter cbf, DataOutputStream dos)
+ throws IOException
+ {
+ /* write the size of the BloomFilter */
+ dos.writeInt(cbf.size());
+ /* write the number of hash functions used */
+ dos.writeInt(cbf.hashes());
+
+ byte[] filter = cbf.filter();
+ /* write length of the filter */
+ dos.writeInt(filter.length);
+ dos.write(filter);
+ }
+
+ public BloomFilter.CountingBloomFilter deserialize(DataInputStream dis) throws IOException
+ {
+ /* read the size of the bloom filter */
+ int size = dis.readInt();
+ /* read the number of hash functions */
+ int hashes = dis.readInt();
+ /* read the length of the filter */
+ int length = dis.readInt();
+ byte[] filter = new byte[length];
+ dis.readFully(filter);
+ return new BloomFilter.CountingBloomFilter(size, hashes, filter);
+ }
+}
+
+interface ISimpleHash
+{
+ public int hash(String str);
+}
+
+class RSHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int b = 378551;
+ int a = 63689;
+ int hash = 0;
+
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = hash * a + str.charAt(i);
+ a = a * b;
+ }
+ return hash;
+ }
+}
+
+class JSHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = 1315423911;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash ^= ((hash << 5) + str.charAt(i) + (hash >> 2));
+ }
+ return hash;
+ }
+}
+
+class PJWHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int bitsInUnsignedInt = (4 * 8);
+ int threeQuarters = (bitsInUnsignedInt * 3) / 4;
+ int oneEighth = bitsInUnsignedInt / 8;
+ int highBits = (0xFFFFFFFF) << (bitsInUnsignedInt - oneEighth);
+ int hash = 0;
+ int test = 0;
+
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = (hash << oneEighth) + str.charAt(i);
+
+ if ((test = hash & highBits) != 0)
+ {
+ hash = ((hash ^ (test >> threeQuarters)) & (~highBits));
+ }
+ }
+ return hash;
+ }
+}
+
+class ELFHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = 0;
+ int x = 0;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = (hash << 4) + str.charAt(i);
+
+ if ((x = hash & 0xF0000000) != 0)
+ {
+ hash ^= (x >> 24);
+ }
+ hash &= ~x;
+ }
+ return hash;
+ }
+}
+
+class BKDRHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int seed = 131; // 31 131 1313 13131 131313 etc..
+ int hash = 0;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = (hash * seed) + str.charAt(i);
+ }
+ return hash;
+ }
+}
+
+class SDBMHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = 0;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = str.charAt(i) + (hash << 6) + (hash << 16) - hash;
+ }
+ return hash;
+ }
+}
+
+class DJBHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = 5381;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = ((hash << 5) + hash) + str.charAt(i);
+ }
+ return hash;
+ }
+}
+
+class DEKHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = str.length();
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = ((hash << 5) ^ (hash >> 27)) ^ str.charAt(i);
+ }
+ return hash;
+ }
+}
+
+class BPHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = 0;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash = hash << 7 ^ str.charAt(i);
+ }
+ return hash;
+ }
+}
+
+class FNVHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int fnv_prime = 0x811C9DC5;
+ int hash = 0;
+ for (int i = 0; i < str.length(); i++)
+ {
+ hash *= fnv_prime;
+ hash ^= str.charAt(i);
+ }
+ return hash;
+ }
+}
+
+class APHash implements ISimpleHash
+{
+ public int hash(String str)
+ {
+ int hash = 0xAAAAAAAA;
+ for (int i = 0; i < str.length(); i++)
+ {
+ if ((i & 1) == 0)
+ {
+ hash ^= ((hash << 7) ^ str.charAt(i) ^ (hash >> 3));
+ }
+ else
+ {
+ hash ^= (~((hash << 11) ^ str.charAt(i) ^ (hash >> 5)));
+ }
+ }
+ return hash;
+ }
+}
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/Cachetable.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/Cachetable.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/Cachetable.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/Cachetable.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,219 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.log4j.Logger;
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class Cachetable<K,V> implements ICachetable<K,V>
+{
+ private class CacheableObject<V>
+ {
+ private V value_;
+ private long age_;
+
+ CacheableObject(V o)
+ {
+ value_ = o;
+ age_ = System.currentTimeMillis();
+ }
+
+ public boolean equals(Object o)
+ {
+ return value_.equals(o);
+ }
+
+ public int hashCode()
+ {
+ return value_.hashCode();
+ }
+
+ V getValue()
+ {
+ return value_;
+ }
+
+ boolean isReadyToDie(long expiration)
+ {
+ return ( (System.currentTimeMillis() - age_) > expiration );
+ }
+ }
+
+ private class CacheMonitor extends TimerTask
+ {
+ private long expiration_;
+
+ CacheMonitor(long expiration)
+ {
+ expiration_ = expiration;
+ }
+
+ public void run()
+ {
+ Map<K,V> expungedValues = new HashMap<K,V>();
+ synchronized(cache_)
+ {
+ Enumeration<K> e = cache_.keys();
+ while( e.hasMoreElements() )
+ {
+ K key = e.nextElement();
+ CacheableObject<V> co = cache_.get(key);
+ if ( co != null && co.isReadyToDie(expiration_) )
+ {
+ V v = co.getValue();
+ if(null != v)
+ expungedValues.put(key, v);
+ cache_.remove(key);
+ }
+ }
+ }
+
+ /* Calling the hooks on the keys that have been expunged */
+ Set<K> keys = expungedValues.keySet();
+ for ( K key : keys )
+ {
+ V value = expungedValues.get(key);
+ ICacheExpungeHook<K,V> hook = hooks_.remove(key);
+ try
+ {
+ if ( hook != null )
+ {
+ hook.callMe(key, value);
+ }
+ else if ( globalHook_ != null )
+ {
+ globalHook_.callMe(key, value);
+ }
+ }
+ catch(Throwable e)
+ {
+ logger_.info(LogUtil.throwableToString(e));
+ }
+ }
+ expungedValues.clear();
+ }
+ }
+
+ private ICacheExpungeHook<K,V> globalHook_;
+ private Hashtable<K, CacheableObject<V>> cache_;
+ private Map<K, ICacheExpungeHook<K,V>> hooks_;
+ private Timer timer_;
+ private static int counter_ = 0;
+ private static Logger logger_ = Logger.getLogger(Cachetable.class);
+
+ private void init(long expiration)
+ {
+ if ( expiration <= 0 )
+ throw new IllegalArgumentException("Argument specified must be a positive number");
+
+ cache_ = new Hashtable<K, CacheableObject<V>>();
+ hooks_ = new Hashtable<K, ICacheExpungeHook<K,V>>();
+ timer_ = new Timer("CACHETABLE-TIMER-" + (++counter_), true);
+ timer_.schedule(new CacheMonitor(expiration), expiration, expiration);
+ }
+
+ /*
+ * Specify the TTL for objects in the cache
+ * in milliseconds.
+ */
+ public Cachetable(long expiration)
+ {
+ init(expiration);
+ }
+
+ /*
+ * Specify the TTL for objects in the cache
+ * in milliseconds and a global expunge hook. If
+ * a key has a key-specific hook installed invoke that
+ * instead.
+ */
+ public Cachetable(long expiration, ICacheExpungeHook<K,V> global)
+ {
+ init(expiration);
+ globalHook_ = global;
+ }
+
+ public void shutdown()
+ {
+ timer_.cancel();
+ }
+
+ public void put(K key, V value)
+ {
+ cache_.put(key, new CacheableObject<V>(value));
+ }
+
+ public void put(K key, V value, ICacheExpungeHook<K,V> hook)
+ {
+ put(key, value);
+ hooks_.put(key, hook);
+ }
+
+ public V get(K key)
+ {
+ V result = null;
+ CacheableObject<V> co = cache_.get(key);
+ if ( co != null )
+ {
+ result = co.getValue();
+ }
+ return result;
+ }
+
+ public V remove(K key)
+ {
+ CacheableObject<V> co = cache_.remove(key);
+ V result = null;
+ if ( co != null )
+ {
+ result = co.getValue();
+ }
+ return result;
+ }
+
+ public int size()
+ {
+ return cache_.size();
+ }
+
+ public boolean containsKey(K key)
+ {
+ return cache_.containsKey(key);
+ }
+
+ public boolean containsValue(V value)
+ {
+ return cache_.containsValue( new CacheableObject<V>(value) );
+ }
+
+ public boolean isEmpty()
+ {
+ return cache_.isEmpty();
+ }
+
+ public Set<K> keySet()
+ {
+ return cache_.keySet();
+ }
+}
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/FBUtilities.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/FBUtilities.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/FBUtilities.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/FBUtilities.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,370 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils;
+
+import java.util.*;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.zip.Deflater;
+import java.util.zip.DataFormatException;
+import java.util.zip.Inflater;
+import java.security.MessageDigest;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.io.*;
+import java.net.*;
+import java.nio.channels.SocketChannel;
+import java.nio.ByteBuffer;
+import java.math.BigInteger;
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class FBUtilities
+{
+
+ private static InetAddress localInetAddress_;
+ private static String host_;
+
+ public static String getTimestamp()
+ {
+ Date date = new Date();
+ DateFormat df = new SimpleDateFormat("MM/dd/yyyy HH:mm:ss");
+ return df.format(date);
+ }
+
+ public static String getTimestamp(long value)
+ {
+ Date date = new Date(value);
+ DateFormat df = new SimpleDateFormat("MM/dd/yyyy HH:mm:ss");
+ return df.format(date);
+ }
+
+ public static int getBits(int x, int p, int n)
+ {
+ return ( x >>> (p + 1 - n)) & ~(~0 << n);
+ }
+
+ public static String getCurrentThreadStackTrace()
+ {
+ Throwable throwable = new Throwable();
+ StackTraceElement[] ste = throwable.getStackTrace();
+ StringBuffer sbuf = new StringBuffer();
+
+ for ( int i = ste.length - 1; i > 0; --i )
+ {
+ sbuf.append(ste[i].getClassName() + "." + ste[i].getMethodName());
+ sbuf.append("/");
+ }
+ sbuf.deleteCharAt(sbuf.length() - 1);
+ return sbuf.toString();
+ }
+
+ public static String[] strip(String string, String token)
+ {
+ StringTokenizer st = new StringTokenizer(string, token);
+ List<String> result = new ArrayList<String>();
+ while ( st.hasMoreTokens() )
+ {
+ result.add( (String)st.nextElement() );
+ }
+ return result.toArray( new String[0] );
+ }
+
+ public static byte[] serializeToStream(Object o)
+ {
+ ByteArrayOutputStream bos = new ByteArrayOutputStream();
+ byte[] bytes = new byte[0];
+ try
+ {
+ ObjectOutputStream oos = new ObjectOutputStream(bos);
+ oos.writeObject(o);
+ oos.flush();
+ bytes = bos.toByteArray();
+ oos.close();
+ bos.close();
+ }
+ catch ( IOException e )
+ {
+ LogUtil.getLogger(FBUtilities.class.getName()).info( LogUtil.throwableToString(e) );
+ }
+ return bytes;
+ }
+
+ public static Object deserializeFromStream(byte[] bytes)
+ {
+ Object o = null;
+ ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
+ try
+ {
+ ObjectInputStream ois = new ObjectInputStream(bis);
+ try
+ {
+ o = ois.readObject();
+ }
+ catch ( ClassNotFoundException e )
+ {
+ }
+ ois.close();
+ bis.close();
+ }
+ catch ( IOException e )
+ {
+ LogUtil.getLogger(FBUtilities.class.getName()).info( LogUtil.throwableToString(e) );
+ }
+ return o;
+ }
+
+ public static InetAddress getLocalAddress() throws UnknownHostException
+ {
+ if ( localInetAddress_ == null )
+ localInetAddress_ = InetAddress.getLocalHost();
+ return localInetAddress_;
+ }
+
+ public static String getLocalHostName() throws UnknownHostException
+ {
+ if ( host_ == null )
+ {
+ host_ = getLocalAddress().getHostName();
+ }
+ return host_;
+ }
+
+ public static String getHostName() throws UnknownHostException
+ {
+ return getLocalAddress().getCanonicalHostName();
+ }
+
+ public static boolean isHostLocalHost(InetAddress host)
+ {
+ try {
+ return getLocalAddress().equals(host);
+ }
+ catch ( UnknownHostException e )
+ {
+ return false;
+ }
+ }
+
+ public static byte[] toByteArray(int i)
+ {
+ byte[] bytes = new byte[4];
+ bytes[0] = (byte)( ( i >>> 24 ) & 0xFF);
+ bytes[1] = (byte)( ( i >>> 16 ) & 0xFF);
+ bytes[2] = (byte)( ( i >>> 8 ) & 0xFF);
+ bytes[3] = (byte)( i & 0xFF );
+ return bytes;
+ }
+
+ public static int byteArrayToInt(byte[] bytes)
+ {
+ return byteArrayToInt(bytes, 0);
+ }
+
+ public static int byteArrayToInt(byte[] bytes, int offset)
+ {
+ if ( bytes.length - offset < 4 )
+ {
+ throw new IllegalArgumentException("An integer must be 4 bytes in size.");
+ }
+ int n = 0;
+ for ( int i = 0; i < 4; ++i )
+ {
+ n <<= 8;
+ n |= bytes[offset + i] & 0xFF;
+ }
+ return n;
+ }
+
+ public static boolean isEqualBits(byte[] bytes1, byte[] bytes2)
+ {
+ return 0 == compareByteArrays(bytes1, bytes2);
+ }
+
+ public static int compareByteArrays(byte[] bytes1, byte[] bytes2){
+ if(null == bytes1){
+ if(null == bytes2) return 0;
+ else return -1;
+ }
+ if(null == bytes2) return 1;
+
+ for(int i = 0; i < bytes1.length && i < bytes2.length; i++){
+ int cmp = compareBytes(bytes1[i], bytes2[i]);
+ if(0 != cmp) return cmp;
+ }
+ if(bytes1.length == bytes2.length) return 0;
+ else return (bytes1.length < bytes2.length)? -1 : 1;
+ }
+
+ public static int compareBytes(byte b1, byte b2){
+ return compareBytes((int)b1, (int)b2);
+ }
+
+ public static int compareBytes(int b1, int b2){
+ int i1 = b1 & 0xFF;
+ int i2 = b2 & 0xFF;
+ if(i1 < i2) return -1;
+ else if(i1 == i2) return 0;
+ else return 1;
+ }
+
+ public static String stackTrace(Throwable e)
+ {
+ StringWriter sw = new StringWriter();
+ PrintWriter pw = new PrintWriter( sw );
+ e.printStackTrace(pw);
+ return sw.toString();
+ }
+
+ public static BigInteger hash(String data)
+ {
+ byte[] result = hash(HashingSchemes.MD5, data.getBytes());
+ BigInteger hash = new BigInteger(result);
+ return hash.abs();
+ }
+
+ public static byte[] hash(String type, byte[] data)
+ {
+ byte[] result = null;
+ try
+ {
+ MessageDigest messageDigest = MessageDigest.getInstance(type);
+ result = messageDigest.digest(data);
+ }
+ catch (Exception e)
+ {
+ LogUtil.getLogger(FBUtilities.class.getName()).debug(LogUtil.throwableToString(e));
+ }
+ return result;
+ }
+
+ public static boolean isEqual(byte[] digestA, byte[] digestB)
+ {
+ return MessageDigest.isEqual(digestA, digestB);
+ }
+
+ // The given bytearray is compressed onto the specifed stream.
+ // The method does not close the stream. The caller will have to do it.
+ public static void compressToStream(byte[] input, ByteArrayOutputStream bos) throws IOException
+ {
+ // Create the compressor with highest level of compression
+ Deflater compressor = new Deflater();
+ compressor.setLevel(Deflater.BEST_COMPRESSION);
+
+ // Give the compressor the data to compress
+ compressor.setInput(input);
+ compressor.finish();
+
+ // Write the compressed data to the stream
+ byte[] buf = new byte[1024];
+ while (!compressor.finished())
+ {
+ int count = compressor.deflate(buf);
+ bos.write(buf, 0, count);
+ }
+ }
+
+
+ public static byte[] compress(byte[] input) throws IOException
+ {
+ // Create an expandable byte array to hold the compressed data.
+ // You cannot use an array that's the same size as the orginal because
+ // there is no guarantee that the compressed data will be smaller than
+ // the uncompressed data.
+ ByteArrayOutputStream bos = new ByteArrayOutputStream(input.length);
+ compressToStream(input,bos);
+ bos.close();
+
+ // Get the compressed data
+ return bos.toByteArray();
+ }
+
+
+ public static byte[] decompress(byte[] compressedData, int off, int len) throws IOException, DataFormatException
+ {
+ // Create the decompressor and give it the data to compress
+ Inflater decompressor = new Inflater();
+ decompressor.setInput(compressedData, off, len);
+
+ // Create an expandable byte array to hold the decompressed data
+ ByteArrayOutputStream bos = new ByteArrayOutputStream(compressedData.length);
+
+ // Decompress the data
+ byte[] buf = new byte[1024];
+ while (!decompressor.finished())
+ {
+ int count = decompressor.inflate(buf);
+ bos.write(buf, 0, count);
+ }
+ bos.close();
+
+ // Get the decompressed data
+ return bos.toByteArray();
+ }
+
+ public static byte[] decompress(byte[] compressedData) throws IOException, DataFormatException
+ {
+ return decompress(compressedData, 0, compressedData.length);
+ }
+
+ public static byte[] xor(byte[] b1, byte[] b2)
+ {
+ byte[] bLess = null;
+ byte[] bMore = null;
+
+ if(b1.length > b2.length)
+ {
+ bLess = b2;
+ bMore = b1;
+ }
+ else
+ {
+ bLess = b1;
+ bMore = b2;
+ }
+
+ for(int i = 0 ; i< bLess.length; i++ )
+ {
+ bMore[i] = (byte)(bMore[i] ^ bLess[i]);
+ }
+
+ return bMore;
+ }
+
+ public static int getUTF8Length(String string)
+ {
+ /*
+ * We store the string as UTF-8 encoded, so when we calculate the length, it
+ * should be converted to UTF-8.
+ */
+ String utfName = string;
+ int length = utfName.length();
+ try
+ {
+ //utfName = new String(string.getBytes("UTF-8"));
+ length = string.getBytes("UTF-8").length;
+ }
+ catch (UnsupportedEncodingException e)
+ {
+ LogUtil.getLogger(FBUtilities.class.getName()).info(LogUtil.throwableToString(e));
+ }
+
+ return length;
+ }
+}
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHash.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHash.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHash.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHash.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,384 @@
+package org.apache.cassandra.utils;
+
+import java.util.Random;
+
+
+/**
+ * Base class for hashtables that use open addressing to resolve collisions.
+ *
+ * @author Avinash Lakshman
+ *
+ */
+
+abstract public class FastHash implements Cloneable
+{
+ /** the current number of occupied slots in the hash. */
+ protected transient int size_;
+
+ /** the current number of free slots in the hash. */
+ protected transient int free_;
+
+ /** the load above which rehashing occurs. */
+ protected static final float DEFAULT_LOAD_FACTOR = 0.5f;
+
+ /**
+ * the default initial capacity for the hash table. This is one less than a
+ * prime value because one is added to it when searching for a prime
+ * capacity to account for the free slot required by open addressing. Thus,
+ * the real default capacity is 11.
+ */
+ protected static final int DEFAULT_INITIAL_CAPACITY = 10;
+
+ /**
+ * Determines how full the internal table can become before rehashing is
+ * required. This must be a value in the range: 0.0 < loadFactor < 1.0. The
+ * default value is 0.5, which is about as large as you can get in open
+ * addressing without hurting performance. Cf. Knuth, Volume 3., Chapter 6.
+ */
+ protected float loadFactor_;
+
+ /**
+ * The maximum number of elements allowed without allocating more space.
+ */
+ protected int maxSize_;
+
+ /**
+ * The number of removes that should be performed before an auto-compaction
+ * occurs.
+ */
+ protected int autoCompactRemovesRemaining_;
+
+ /**
+ * The auto-compaction factor for the table.
+ *
+ * @see #setAutoCompactionFactor
+ */
+ protected float autoCompactionFactor_;
+
+ /**
+ * @see
+ */
+ private boolean autoCompactTemporaryDisable_ = false;
+
+ /**
+ * Creates a new <code>THash</code> instance with the default capacity and
+ * load factor.
+ */
+ public FastHash()
+ {
+ this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR);
+ }
+
+ /**
+ * Creates a new <code>THash</code> instance with a prime capacity at or
+ * near the specified capacity and with the default load factor.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ */
+ public FastHash(int initialCapacity)
+ {
+ this(initialCapacity, DEFAULT_LOAD_FACTOR);
+ }
+
+ /**
+ * Creates a new <code>THash</code> instance with a prime capacity at or
+ * near the minimum needed to hold <tt>initialCapacity</tt> elements with
+ * load factor <tt>loadFactor</tt> without triggering a rehash.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ * @param loadFactor
+ * a <code>float</code> value
+ */
+ public FastHash(int initialCapacity, float loadFactor)
+ {
+ super();
+ loadFactor_ = loadFactor;
+
+ // Through testing, the load factor (especially the default load factor)
+ // has been
+ // found to be a pretty good starting auto-compaction factor.
+ autoCompactionFactor_ = loadFactor;
+
+ setUp((int) Math.ceil(initialCapacity / loadFactor));
+ }
+
+ public Object clone()
+ {
+ try
+ {
+ return super.clone();
+ }
+ catch (CloneNotSupportedException cnse)
+ {
+ return null; // it's supported
+ }
+ }
+
+ /**
+ * Tells whether this set is currently holding any elements.
+ *
+ * @return a <code>boolean</code> value
+ */
+ public boolean isEmpty()
+ {
+ return 0 == size_;
+ }
+
+ /**
+ * Returns the number of distinct elements in this collection.
+ *
+ * @return an <code>int</code> value
+ */
+ public int size()
+ {
+ return size_;
+ }
+
+ /**
+ * @return the current physical capacity of the hash table.
+ */
+ abstract protected int capacity();
+
+ /**
+ * Ensure that this hashtable has sufficient capacity to hold
+ * <tt>desiredCapacity<tt> <b>additional</b> elements without
+ * requiring a rehash. This is a tuning method you can call
+ * before doing a large insert.
+ *
+ * @param desiredCapacity an <code>int</code> value
+ */
+ public void ensureCapacity(int desiredCapacity)
+ {
+ if (desiredCapacity > (maxSize_ - size()))
+ {
+ rehash(PrimeFinder.nextPrime((int) Math.ceil(desiredCapacity
+ + size() / loadFactor_) + 1));
+ computeMaxSize(capacity());
+ }
+ }
+
+ /**
+ * Compresses the hashtable to the minimum prime size (as defined by
+ * PrimeFinder) that will hold all of the elements currently in the table.
+ * If you have done a lot of <tt>remove</tt> operations and plan to do a
+ * lot of queries or insertions or iteration, it is a good idea to invoke
+ * this method. Doing so will accomplish two things:
+ *
+ * <ol>
+ * <li> You'll free memory allocated to the table but no longer needed
+ * because of the remove()s.</li>
+ *
+ * <li> You'll get better query/insert/iterator performance because there
+ * won't be any <tt>REMOVED</tt> slots to skip over when probing for
+ * indices in the table.</li>
+ * </ol>
+ */
+ public void compact()
+ {
+ // need at least one free spot for open addressing
+ rehash(PrimeFinder.nextPrime((int) Math.ceil(size() / loadFactor_) + 1));
+ computeMaxSize(capacity());
+
+ // If auto-compaction is enabled, re-determine the compaction interval
+ if (autoCompactionFactor_ != 0)
+ {
+ computeNextAutoCompactionAmount(size());
+ }
+ }
+
+ /**
+ * The auto-compaction factor controls whether and when a table performs a
+ * {@link #compact} automatically after a certain number of remove
+ * operations. If the value is non-zero, the number of removes that need to
+ * occur for auto-compaction is the size of table at the time of the
+ * previous compaction (or the initial capacity) multiplied by this factor.
+ * <p>
+ * Setting this value to zero will disable auto-compaction.
+ */
+ public void setAutoCompactionFactor(float factor)
+ {
+ if (factor < 0)
+ {
+ throw new IllegalArgumentException("Factor must be >= 0: " + factor);
+ }
+
+ autoCompactionFactor_ = factor;
+ }
+
+ /**
+ * @see #setAutoCompactionFactor
+ */
+ public float getAutoCompactionFactor()
+ {
+ return autoCompactionFactor_;
+ }
+
+ /**
+ * This simply calls {@link #compact compact}. It is included for symmetry
+ * with other collection classes. Note that the name of this method is
+ * somewhat misleading (which is why we prefer <tt>compact</tt>) as the
+ * load factor may require capacity above and beyond the size of this
+ * collection.
+ *
+ * @see #compact
+ */
+ public final void trimToSize()
+ {
+ compact();
+ }
+
+ /**
+ * Delete the record at <tt>index</tt>. Reduces the size of the
+ * collection by one.
+ *
+ * @param index
+ * an <code>int</code> value
+ */
+ protected void removeAt(int index)
+ {
+ size_--;
+
+ // If auto-compaction is enabled, see if we need to compact
+ if (autoCompactionFactor_ != 0)
+ {
+ autoCompactRemovesRemaining_--;
+
+ if (!autoCompactTemporaryDisable_
+ && autoCompactRemovesRemaining_ <= 0)
+ {
+ // Do the compact
+ // NOTE: this will cause the next compaction interval to be
+ // calculated
+ compact();
+ }
+ }
+ }
+
+ /**
+ * Empties the collection.
+ */
+ public void clear()
+ {
+ size_ = 0;
+ free_ = capacity();
+ }
+
+ /**
+ * initializes the hashtable to a prime capacity which is at least
+ * <tt>initialCapacity + 1</tt>.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ * @return the actual capacity chosen
+ */
+ protected int setUp(int initialCapacity)
+ {
+ int capacity;
+
+ capacity = PrimeFinder.nextPrime(initialCapacity);
+ computeMaxSize(capacity);
+ computeNextAutoCompactionAmount(initialCapacity);
+
+ return capacity;
+ }
+
+ /**
+ * Rehashes the set.
+ *
+ * @param newCapacity
+ * an <code>int</code> value
+ */
+ protected abstract void rehash(int newCapacity);
+
+ /**
+ * Temporarily disables auto-compaction. MUST be followed by calling
+ * {@link #reenableAutoCompaction}.
+ */
+ protected void tempDisableAutoCompaction()
+ {
+ autoCompactTemporaryDisable_ = true;
+ }
+
+ /**
+ * Re-enable auto-compaction after it was disabled via
+ * {@link #tempDisableAutoCompaction()}.
+ *
+ * @param check_for_compaction
+ * True if compaction should be performed if needed before
+ * returning. If false, no compaction will be performed.
+ */
+ protected void reenableAutoCompaction(boolean check_for_compaction)
+ {
+ autoCompactTemporaryDisable_ = false;
+
+ if (check_for_compaction && autoCompactRemovesRemaining_ <= 0
+ && autoCompactionFactor_ != 0)
+ {
+
+ // Do the compact
+ // NOTE: this will cause the next compaction interval to be
+ // calculated
+ compact();
+ }
+ }
+
+ /**
+ * Computes the values of maxSize. There will always be at least one free
+ * slot required.
+ *
+ * @param capacity
+ * an <code>int</code> value
+ */
+ private final void computeMaxSize(int capacity)
+ {
+ // need at least one free slot for open addressing
+ maxSize_ = Math.min(capacity - 1, (int) Math.floor(capacity
+ * loadFactor_));
+ free_ = capacity - size_; // reset the free element count
+ }
+
+ /**
+ * Computes the number of removes that need to happen before the next
+ * auto-compaction will occur.
+ */
+ private void computeNextAutoCompactionAmount(int size)
+ {
+ if (autoCompactionFactor_ != 0)
+ {
+ autoCompactRemovesRemaining_ = Math.round(size
+ * autoCompactionFactor_);
+ }
+ }
+
+ /**
+ * After an insert, this hook is called to adjust the size/free values of
+ * the set and to perform rehashing if necessary.
+ */
+ protected final void postInsertHook(boolean usedFreeSlot)
+ {
+ if (usedFreeSlot)
+ {
+ free_--;
+ }
+
+ // rehash whenever we exhaust the available space in the table
+ if (++size_ > maxSize_ || free_ == 0)
+ {
+ // choose a new capacity suited to the new state of the table
+ // if we've grown beyond our maximum size, double capacity;
+ // if we've exhausted the free spots, rehash to the same capacity,
+ // which will free up any stale removed slots for reuse.
+ int newCapacity = size_ > maxSize_ ? PrimeFinder
+ .nextPrime(capacity() << 1) : capacity();
+ rehash(newCapacity);
+ computeMaxSize(capacity());
+ }
+ }
+
+ protected int calculateGrownCapacity()
+ {
+ return capacity() << 1;
+ }
+}// THash
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHashMap.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHashMap.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHashMap.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastHashMap.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,583 @@
+package org.apache.cassandra.utils;
+
+import java.io.*;
+import java.util.*;
+
+/**
+ * An implementation of the Map interface which uses an open addressed hash
+ * table to store its contents
+ * @author Avinash Lakshman
+ */
+public class FastHashMap<K, V> extends FastObjectHash<K> implements Map<K, V>, Serializable
+{
+ static final long serialVersionUID = 1L;
+
+ /** the values of the map */
+ protected transient V[] values_;
+
+ /**
+ * Creates a new <code>FastHashMap</code> instance with the default capacity
+ * and load factor.
+ */
+ public FastHashMap()
+ {
+ super();
+ }
+
+ /**
+ * Creates a new <code>FastHashMap</code> instance with a prime capacity
+ * equal to or greater than <tt>initialCapacity</tt> and with the default
+ * load factor.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ */
+ public FastHashMap(int initialCapacity)
+ {
+ super(initialCapacity);
+ }
+
+ /**
+ * Creates a new <code>FastHashMap</code> instance with a prime capacity
+ * equal to or greater than <tt>initialCapacity</tt> and with the
+ * specified load factor.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ * @param loadFactor
+ * a <code>float</code> value
+ */
+ public FastHashMap(int initialCapacity, float loadFactor)
+ {
+ super(initialCapacity, loadFactor);
+ }
+
+ /**
+ * Creates a new <code>FastHashMap</code> instance which contains the
+ * key/value pairs in <tt>map</tt>.
+ *
+ * @param map
+ * a <code>Map</code> value
+ */
+ public FastHashMap(Map<K, V> map)
+ {
+ this(map.size());
+ putAll(map);
+ }
+
+ /**
+ * @return a shallow clone of this collection
+ */
+ public FastHashMap<K, V> clone()
+ {
+ FastHashMap<K, V> m = (FastHashMap<K, V>) super.clone();
+ m.values_ = this.values_.clone();
+ return m;
+ }
+
+ /**
+ * initialize the value array of the map.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ * @return an <code>int</code> value
+ */
+ protected int setUp(int initialCapacity)
+ {
+ int capacity;
+
+ capacity = super.setUp(initialCapacity);
+ values_ = (V[]) new Object[capacity];
+ return capacity;
+ }
+
+ void addEntry(Object key, int index)
+ {
+ }
+
+ void removeEntry(Object key, int index)
+ {
+ }
+
+ /**
+ * Inserts a key/value pair into the map.
+ *
+ * @param key
+ * an <code>Object</code> value
+ * @param value
+ * an <code>Object</code> value
+ * @return the previous value associated with <tt>key</tt>, or null if
+ * none was found.
+ */
+ public V put(K key, V value)
+ {
+ V previous = null;
+ Object oldKey;
+ int index = insertionIndex(key);
+ boolean isNewMapping = true;
+ if (index < 0)
+ {
+ index = -index - 1;
+ previous = values_[index];
+ isNewMapping = false;
+ }
+ oldKey = set_[index];
+
+ if ( oldKey == FREE )
+ {
+ /* This is used as a hook to process new put() operations */
+ addEntry(key, index);
+ }
+
+ set_[index] = key;
+ values_[index] = value;
+ if (isNewMapping)
+ {
+ postInsertHook(oldKey == FREE);
+ }
+
+ return previous;
+ }
+
+ /**
+ * rehashes the map to the new capacity.
+ *
+ * @param newCapacity
+ * an <code>int</code> value
+ */
+ protected void rehash(int newCapacity)
+ {
+ int oldCapacity = set_.length;
+ Object oldKeys[] = set_;
+ V oldVals[] = values_;
+
+ set_ = new Object[newCapacity];
+ Arrays.fill(set_, FREE);
+ values_ = (V[]) new Object[newCapacity];
+
+ for (int i = oldCapacity; i-- > 0;)
+ {
+ if (oldKeys[i] != FREE && oldKeys[i] != REMOVED)
+ {
+ Object o = oldKeys[i];
+ int index = insertionIndex((K) o);
+ if (index < 0)
+ {
+ throwObjectContractViolation(set_[(-index - 1)], o);
+ }
+ set_[index] = o;
+ values_[index] = oldVals[i];
+ }
+ }
+ }
+
+ /**
+ * retrieves the value for <tt>key</tt>
+ *
+ * @param key
+ * an <code>Object</code> value
+ * @return the value of <tt>key</tt> or null if no such mapping exists.
+ */
+ public V get(Object key)
+ {
+ int index = index((K) key);
+ return index < 0 ? null : values_[index];
+ }
+
+ /**
+ * Empties the map.
+ *
+ */
+ public void clear()
+ {
+ if (size() == 0)
+ return; // optimization
+
+ super.clear();
+ Object[] keys = set_;
+ V[] vals = values_;
+
+ for (int i = keys.length; i-- > 0;)
+ {
+ keys[i] = FREE;
+ vals[i] = null;
+ }
+ }
+
+ /**
+ * Deletes a key/value pair from the map.
+ *
+ * @param key an <code>Object</code> value
+ * @return an <code>Object</code> value
+ */
+ public V remove(Object key)
+ {
+ V prev = null;
+ int index = index((K)key);
+ if (index >= 0)
+ {
+ prev = values_[index];
+ /* clear key,state; adjust size */
+ removeAt(index);
+ /* This is used as hook to process deleted items */
+ removeEntry(key, index);
+ }
+ return prev;
+ }
+
+ /**
+ * removes the mapping at <tt>index</tt> from the map.
+ *
+ * @param index an <code>int</code> value
+ */
+ protected void removeAt(int index)
+ {
+ values_[index] = null;
+ /* clear key, state; adjust size */
+ super.removeAt(index);
+ }
+
+ /**
+ * Returns a view on the values of the map.
+ *
+ * @return a <code>Collection</code> value
+ */
+ public Collection<V> values()
+ {
+ return Arrays.asList(values_);
+ }
+
+ /**
+ * returns a Set view on the keys of the map.
+ *
+ * @return a <code>Set</code> value
+ */
+ public Set<K> keySet()
+ {
+ return new KeyView();
+ }
+
+ /**
+ * Returns a Set view on the entries of the map.
+ *
+ * @return a <code>Set</code> value
+ */
+ public Set<Map.Entry<K, V>> entrySet()
+ {
+ throw new UnsupportedOperationException(
+ "This operation is currently not supported.");
+ }
+
+ /**
+ * checks for the presence of <tt>val</tt> in the values of the map.
+ *
+ * @param val
+ * an <code>Object</code> value
+ * @return a <code>boolean</code> value
+ */
+ public boolean containsValue(Object val)
+ {
+ Object[] set = set_;
+ V[] vals = values_;
+
+ // special case null values so that we don't have to
+ // perform null checks before every call to equals()
+ if (null == val)
+ {
+ for (int i = vals.length; i-- > 0;)
+ {
+ if ((set[i] != FREE && set[i] != REMOVED) && val == vals[i])
+ {
+ return true;
+ }
+ }
+ }
+ else
+ {
+ for (int i = vals.length; i-- > 0;)
+ {
+ if ((set[i] != FREE && set[i] != REMOVED)
+ && (val == vals[i] || val.equals(vals[i])))
+ {
+ return true;
+ }
+ }
+ } // end of else
+ return false;
+ }
+
+ /**
+ * checks for the present of <tt>key</tt> in the keys of the map.
+ *
+ * @param key
+ * an <code>Object</code> value
+ * @return a <code>boolean</code> value
+ */
+ public boolean containsKey(Object key)
+ {
+ return contains(key);
+ }
+
+ /**
+ * copies the key/value mappings in <tt>map</tt> into this map.
+ *
+ * @param map
+ * a <code>Map</code> value
+ */
+ public void putAll(Map<? extends K, ? extends V> map)
+ {
+ ensureCapacity(map.size());
+ // could optimize this for cases when map instanceof FastHashMap
+ for (Iterator<? extends Map.Entry<? extends K, ? extends V>> i = map
+ .entrySet().iterator(); i.hasNext();)
+ {
+ Map.Entry<? extends K, ? extends V> e = i.next();
+ put(e.getKey(), e.getValue());
+ }
+ }
+
+ private abstract class MapBackedView<E> extends AbstractSet<E> implements Set<E>, Iterable<E>
+ {
+ public abstract Iterator<E> iterator();
+ public abstract boolean removeElement(E key);
+ public abstract boolean containsElement(E key);
+
+ public boolean contains(Object key)
+ {
+ return containsElement((E) key);
+ }
+
+ public boolean remove(Object o)
+ {
+ return removeElement((E) o);
+ }
+
+ public boolean containsAll(Collection<?> collection)
+ {
+ for (Iterator i = collection.iterator(); i.hasNext();)
+ {
+ if (!contains(i.next()))
+ {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public void clear()
+ {
+ FastHashMap.this.clear();
+ }
+
+ public boolean add(E obj)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ public int size()
+ {
+ return FastHashMap.this.size();
+ }
+
+ public Object[] toArray()
+ {
+ Object[] result = new Object[size()];
+ Iterator e = iterator();
+ for (int i = 0; e.hasNext(); i++)
+ result[i] = e.next();
+ return result;
+ }
+
+ public <T> T[] toArray(T[] a)
+ {
+ int size = size();
+ if (a.length < size)
+ a = (T[]) java.lang.reflect.Array.newInstance(a.getClass().getComponentType(), size);
+
+ Iterator<E> it = iterator();
+ Object[] result = a;
+ for (int i = 0; i < size; i++)
+ {
+ result[i] = it.next();
+ }
+
+ if (a.length > size)
+ {
+ a[size] = null;
+ }
+
+ return a;
+ }
+
+ public boolean isEmpty()
+ {
+ return FastHashMap.this.isEmpty();
+ }
+
+ public boolean addAll(Collection<? extends E> collection)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ public boolean retainAll(Collection<?> collection)
+ {
+ boolean changed = false;
+ Iterator i = iterator();
+ while (i.hasNext())
+ {
+ if (!collection.contains(i.next()))
+ {
+ i.remove();
+ changed = true;
+ }
+ }
+ return changed;
+ }
+ }
+
+ protected class FastHashMapIterator<T> implements Iterator<T>
+ {
+ private int nextIndex_;
+ private int expectedSize_;
+ private FastObjectHash<T> tMap_;
+
+ FastHashMapIterator(FastObjectHash<T> tMap)
+ {
+ nextIndex_ = -1;
+ expectedSize_ = tMap.size();
+ tMap_ = tMap;
+ }
+
+ public boolean hasNext()
+ {
+ return (expectedSize_ > 0);
+ }
+
+ public T next()
+ {
+ moveToNextIndex();
+ int index = nextIndex_;
+ /*
+ * Decrement so that we can track how many
+ * elements we have already looked at.
+ */
+ --expectedSize_;
+ return (T)tMap_.set_[index];
+ }
+
+ private void moveToNextIndex()
+ {
+ int i = nextIndex_ + 1;
+ for ( ; i < tMap_.set_.length; ++i )
+ {
+ if ( tMap_.set_[i].equals(FREE) || tMap_.set_[i].equals(REMOVED) )
+ {
+ continue;
+ }
+ else
+ {
+ break;
+ }
+ }
+ nextIndex_ = i;
+ }
+
+ public void remove()
+ {
+ tMap_.removeAt(nextIndex_);
+ --expectedSize_;
+ }
+ }
+
+ /**
+ * a view onto the keys of the map.
+ */
+ protected class KeyView extends MapBackedView<K>
+ {
+ public Iterator<K> iterator()
+ {
+ return new FastHashMapIterator(FastHashMap.this);
+ }
+
+ public boolean removeElement(K key)
+ {
+ return null != FastHashMap.this.remove(key);
+ }
+
+ public boolean containsElement(K key)
+ {
+ return FastHashMap.this.contains(key);
+ }
+ }
+
+ final class Entry implements Map.Entry<K, V>
+ {
+ private K key;
+ private V val;
+ private final int index;
+
+ Entry(final K key, V value, final int index)
+ {
+ this.key = key;
+ this.val = value;
+ this.index = index;
+ }
+
+ void setKey(K aKey)
+ {
+ this.key = aKey;
+ }
+
+ void setValue0(V aValue)
+ {
+ this.val = aValue;
+ }
+
+ public K getKey()
+ {
+ return key;
+ }
+
+ public V getValue()
+ {
+ return val;
+ }
+
+ public V setValue(V o)
+ {
+ if (values_[index] != val)
+ {
+ throw new ConcurrentModificationException();
+ }
+ values_[index] = o;
+ o = val; // need to return previous value
+ val = o; // update this entry's value, in case
+ // setValue is called again
+ return o;
+ }
+
+ public boolean equals(Object o)
+ {
+ if (o instanceof Map.Entry)
+ {
+ Map.Entry e1 = this;
+ Map.Entry e2 = (Map.Entry) o;
+ return (e1.getKey() == null ? e2.getKey() == null : e1.getKey().equals(e2.getKey()))
+ && (e1.getValue() == null ? e2.getValue() == null : e1.getValue().equals(e2.getValue()));
+ }
+ return false;
+ }
+
+ public int hashCode()
+ {
+ return (getKey() == null ? 0 : getKey().hashCode()) ^ (getValue() == null ? 0 : getValue().hashCode());
+ }
+ }
+
+ public static void main(String[] args) throws Throwable
+ {
+ Map<String, String> map = new FastHashMap<String, String>();
+ map.put("Avinash", "Avinash");
+ map.put("Avinash", "Srinivas");
+ }
+}
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastLinkedHashMap.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastLinkedHashMap.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastLinkedHashMap.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastLinkedHashMap.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils;
+
+public class FastLinkedHashMap<K,V> extends FastHashMap<K,V>
+{
+
+}
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastObjectHash.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastObjectHash.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastObjectHash.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/FastObjectHash.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,306 @@
+package org.apache.cassandra.utils;
+
+import java.util.Arrays;
+
+/**
+ * An open addressed hashing implementation for Object types.
+ *
+ * @author Avinash Lakshman
+ */
+abstract public class FastObjectHash<T> extends FastHash
+{
+ static final long serialVersionUID = -3461112548087185871L;
+
+ /** the set of Objects */
+ protected transient Object[] set_;
+ protected static final Object REMOVED = new Object(), FREE = new Object();
+
+ /**
+ * Creates a new <code>TObjectHash</code> instance with the default
+ * capacity and load factor.
+ */
+ public FastObjectHash()
+ {
+ super();
+ }
+
+ /**
+ * Creates a new <code>TObjectHash</code> instance whose capacity is the
+ * next highest prime above <tt>initialCapacity + 1</tt> unless that value
+ * is already prime.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ */
+ public FastObjectHash(int initialCapacity)
+ {
+ super(initialCapacity);
+ }
+
+ /**
+ * Creates a new <code>TObjectHash</code> instance with a prime value at
+ * or near the specified capacity and load factor.
+ *
+ * @param initialCapacity
+ * used to find a prime capacity for the table.
+ * @param loadFactor
+ * used to calculate the threshold over which rehashing takes
+ * place.
+ */
+ public FastObjectHash(int initialCapacity, float loadFactor)
+ {
+ super(initialCapacity, loadFactor);
+ }
+
+ /**
+ * @return a shallow clone of this collection
+ */
+ public FastObjectHash<T> clone()
+ {
+ FastObjectHash<T> h = (FastObjectHash<T>) super.clone();
+ h.set_ = (Object[]) this.set_.clone();
+ return h;
+ }
+
+ /**
+ * This method is invoked every time a key is
+ * added into the Map.
+ * @param key key that is inserted
+ * @param index index position of the key being
+ * inserted
+ */
+ abstract void addEntry(Object key, int index);
+
+ /**
+ * This method is invoked every time a key is
+ * deleted from the Map.
+ * @param key key being deleted
+ * @param index index position of the key being
+ * deleted
+ */
+ abstract void removeEntry(Object key, int index);
+
+ protected int capacity()
+ {
+ return set_.length;
+ }
+
+ protected void removeAt(int index)
+ {
+ set_[index] = REMOVED;
+ super.removeAt(index);
+ }
+
+ /**
+ * initializes the Object set of this hash table.
+ *
+ * @param initialCapacity
+ * an <code>int</code> value
+ * @return an <code>int</code> value
+ */
+ protected int setUp(int initialCapacity)
+ {
+ int capacity;
+
+ capacity = super.setUp(initialCapacity);
+ set_ = new Object[capacity];
+ Arrays.fill(set_, FREE);
+ return capacity;
+ }
+
+ /**
+ * Searches the set for <tt>obj</tt>
+ *
+ * @param obj
+ * an <code>Object</code> value
+ * @return a <code>boolean</code> value
+ */
+ public boolean contains(Object obj)
+ {
+ return index((T) obj) >= 0;
+ }
+
+ /**
+ * Locates the index of <tt>obj</tt>.
+ *
+ * @param obj
+ * an <code>Object</code> value
+ * @return the index of <tt>obj</tt> or -1 if it isn't in the set.
+ */
+ protected int index(Object obj)
+ {
+ final Object[] set = set_;
+ final int length = set.length;
+ final int hash = obj.hashCode() & 0x7fffffff;
+ int index = hash % length;
+ Object cur = set[index];
+
+ if (cur == FREE)
+ return -1;
+
+ // NOTE: here it has to be REMOVED or FULL (some user-given value)
+ if (cur == REMOVED || cur.equals(obj))
+ {
+ // see Knuth, p. 529
+ final int probe = 1 + (hash % (length - 2));
+
+ while (cur != FREE&& (cur == REMOVED || !cur.equals(obj)))
+ {
+ index -= probe;
+ if (index < 0)
+ {
+ index += length;
+ }
+ cur = set[index];
+ }
+ }
+
+ return cur == FREE ? -1 : index;
+ }
+
+ /**
+ * Locates the index at which <tt>obj</tt> can be inserted. if there is
+ * already a value equal()ing <tt>obj</tt> in the set, returns that
+ * value's index as <tt>-index - 1</tt>.
+ *
+ * @param obj
+ * an <code>Object</code> value
+ * @return the index of a FREE slot at which obj can be inserted or, if obj
+ * is already stored in the hash, the negative value of that index,
+ * minus 1: -index -1.
+ */
+ protected int insertionIndex(T obj)
+ {
+ final Object[] set = set_;
+ final int length = set.length;
+ final int hash = obj.hashCode() & 0x7fffffff;
+ int index = hash % length;
+ Object cur = set[index];
+
+ if (cur == FREE)
+ {
+ return index; // empty, all done
+ }
+ else if (cur != REMOVED && cur.equals(obj))
+ {
+ return -index - 1; // already stored
+ }
+ else
+ { // already FULL or REMOVED, must probe
+ // compute the double hash
+ final int probe = 1 + (hash % (length - 2));
+
+ // if the slot we landed on is FULL (but not removed), probe
+ // until we find an empty slot, a REMOVED slot, or an element
+ // equal to the one we are trying to insert.
+ // finding an empty slot means that the value is not present
+ // and that we should use that slot as the insertion point;
+ // finding a REMOVED slot means that we need to keep searching,
+ // however we want to remember the offset of that REMOVED slot
+ // so we can reuse it in case a "new" insertion (i.e. not an update)
+ // is possible.
+ // finding a matching value means that we've found that our desired
+ // key is already in the table
+ if (cur != REMOVED)
+ {
+ // starting at the natural offset, probe until we find an
+ // offset that isn't full.
+ do
+ {
+ index -= probe;
+ if (index < 0)
+ {
+ index += length;
+ }
+ cur = set[index];
+ }
+ while (cur != FREE && cur != REMOVED
+ && !cur.equals(obj));
+ }
+
+ // if the index we found was removed: continue probing until we
+ // locate a free location or an element which equal()s the
+ // one we have.
+ if (cur == REMOVED)
+ {
+ int firstRemoved = index;
+ while (cur != FREE
+ && (cur == REMOVED || !cur.equals(obj)))
+ {
+ index -= probe;
+ if (index < 0)
+ {
+ index += length;
+ }
+ cur = set[index];
+ }
+ // NOTE: cur cannot == REMOVED in this block
+ return (cur != FREE) ? -index - 1 : firstRemoved;
+ }
+ // if it's full, the key is already stored
+ // NOTE: cur cannot equal REMOVE here (would have retuned already
+ // (see above)
+ return (cur != FREE) ? -index - 1 : index;
+ }
+ }
+
+ /**
+ * This is the default implementation of TObjectHashingStrategy: it
+ * delegates hashing to the Object's hashCode method.
+ *
+ * @param o
+ * for which the hashcode is to be computed
+ * @return the hashCode
+ * @see Object#hashCode()
+ */
+ public final int computeHashCode(T o)
+ {
+ return o == null ? 0 : o.hashCode();
+ }
+
+ /**
+ * This is the default implementation of TObjectHashingStrategy: it
+ * delegates equality comparisons to the first parameter's equals() method.
+ *
+ * @param o1
+ * an <code>Object</code> value
+ * @param o2
+ * an <code>Object</code> value
+ * @return true if the objects are equal
+ * @see Object#equals(Object)
+ */
+ public final boolean equals(T o1, T o2)
+ {
+ return o1 == null ? o2 == null : o1.equals(o2);
+ }
+
+ /**
+ * Convenience methods for subclasses to use in throwing exceptions about
+ * badly behaved user objects employed as keys. We have to throw an
+ * IllegalArgumentException with a rather verbose message telling the user
+ * that they need to fix their object implementation to conform to the
+ * general contract for java.lang.Object.
+ *
+ * @param o1
+ * the first of the equal elements with unequal hash codes.
+ * @param o2
+ * the second of the equal elements with unequal hash codes.
+ * @exception IllegalArgumentException
+ * the whole point of this method.
+ */
+ protected final void throwObjectContractViolation(Object o1, Object o2)
+ throws IllegalArgumentException
+ {
+ throw new IllegalArgumentException(
+ "Equal objects must have equal hashcodes. "
+ + "During rehashing, Trove discovered that "
+ + "the following two objects claim to be "
+ + "equal (as in java.lang.Object.equals()) "
+ + "but their hashCodes (or those calculated by "
+ + "your TObjectHashingStrategy) are not equal."
+ + "This violates the general contract of "
+ + "java.lang.Object.hashCode(). See bullet point two "
+ + "in that method's documentation. " + "object #1 ="
+ + o1 + "; object #2 =" + o2);
+ }
+} // TObjectHash
Added: incubator/cassandra/trunk/src/org/apache/cassandra/utils/FileUtils.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/org/apache/cassandra/utils/FileUtils.java?rev=749218&view=auto
==============================================================================
--- incubator/cassandra/trunk/src/org/apache/cassandra/utils/FileUtils.java (added)
+++ incubator/cassandra/trunk/src/org/apache/cassandra/utils/FileUtils.java Mon Mar 2 07:57:22 2009
@@ -0,0 +1,257 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.utils;
+
+import java.io.*;
+import java.text.DecimalFormat;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
+import org.apache.cassandra.concurrent.ThreadFactoryImpl;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.log4j.Logger;
+
+
+
+/**
+ * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
+ */
+
+public class FileUtils
+{
+ private static Logger logger_ = Logger.getLogger(FileUtils.class);
+ private static final DecimalFormat df_ = new DecimalFormat("#.##");
+ private static final double kb_ = 1024d;
+ private static final double mb_ = 1024*1024d;
+ private static final double gb_ = 1024*1024*1024d;
+ private static final double tb_ = 1024*1024*1024*1024d;
+
+ private static ExecutorService deleter_ = new DebuggableThreadPoolExecutor( 1,
+ 1,
+ Integer.MAX_VALUE,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue<Runnable>(),
+ new ThreadFactoryImpl("FILEUTILS-DELETE-POOL")
+ );
+
+ public static void shutdown()
+ {
+ deleter_.shutdownNow();
+ }
+
+ public static class Deleter implements Runnable
+ {
+ File file_ = null;
+
+ public Deleter(File f)
+ {
+ file_ = f;
+ }
+
+ public void run()
+ {
+ if(file_ == null)
+ return;
+ logger_.info("*** Deleting " + file_.getName() + " ***");
+ if(!file_.delete())
+ {
+ logger_.warn("Warning : Unable to delete file " + file_.getAbsolutePath());
+ }
+ }
+ }
+
+ public static class FileComparator implements Comparator<File>
+ {
+ public int compare(File f, File f2)
+ {
+ return (int)(f.lastModified() - f2.lastModified());
+ }
+
+ public boolean equals(Object o)
+ {
+ if ( !(o instanceof FileComparator) )
+ return false;
+ return true;
+ }
+ }
+
+ public static void createDirectory(String directory) throws IOException
+ {
+ File file = new File(directory);
+ if ( !file.exists() )
+ file.mkdir();
+ }
+
+ public static void createFile(String directory) throws IOException
+ {
+ File file = new File(directory);
+ if ( !file.exists() )
+ file.createNewFile();
+ }
+
+ public static boolean isExists(String filename) throws IOException
+ {
+ File file = new File(filename);
+ return file.exists();
+ }
+
+ public static boolean delete(String file)
+ {
+ File f = new File(file);
+ return f.delete();
+ }
+
+ public static void deleteAsync(String file) throws IOException
+ {
+ File f = new File(file);
+ Runnable deleter = new Deleter(f);
+ deleter_.submit(deleter);
+ }
+
+ public static boolean delete(List<String> files) throws IOException
+ {
+ boolean bVal = true;
+ for ( int i = 0; i < files.size(); ++i )
+ {
+ String file = files.get(i);
+ bVal = delete(file);
+ if (bVal)
+ {
+ logger_.debug("Deleted file " + file);
+ files.remove(i);
+ }
+ }
+ return bVal;
+ }
+
+ public static void delete(File[] files) throws IOException
+ {
+ for ( File file : files )
+ {
+ file.delete();
+ }
+ }
+
+ public static String stringifyFileSize(double value)
+ {
+ double d = 0d;
+ if ( value >= tb_ )
+ {
+ d = value / tb_;
+ String val = df_.format(d);
+ return val + " TB";
+ }
+ else if ( value >= gb_ )
+ {
+ d = value / gb_;
+ String val = df_.format(d);
+ return val + " GB";
+ }
+ else if ( value >= mb_ )
+ {
+ d = value / mb_;
+ String val = df_.format(d);
+ return val + " MB";
+ }
+ else if ( value >= kb_ )
+ {
+ d = value / kb_;
+ String val = df_.format(d);
+ return val + " KB";
+ }
+ else
+ {
+ String val = df_.format(value);
+ return val + " bytes.";
+ }
+ }
+
+ public static double stringToFileSize(String value)
+ {
+ String[] peices = value.split(" ");
+ double d = Double.valueOf(peices[0]);
+ if ( peices[1].equals("TB") )
+ {
+ d *= tb_;
+ }
+ else if ( peices[1].equals("GB") )
+ {
+ d *= gb_;
+ }
+ else if ( peices[1].equals("MB") )
+ {
+ d *= mb_;
+ }
+ else if ( peices[1].equals("KB") )
+ {
+ d *= kb_;
+ }
+ else
+ {
+ d *= 1;
+ }
+ return d;
+ }
+
+ public static long getUsedDiskSpace()
+ {
+ long diskSpace = 0L;
+ String[] directories = DatabaseDescriptor.getAllDataFileLocations();
+ for ( String directory : directories )
+ {
+ File f = new File(directory);
+ File[] files = f.listFiles();
+ for ( File file : files )
+ {
+ diskSpace += file.length();
+ }
+ }
+
+ String value = df_.format(diskSpace);
+ return Long.parseLong(value);
+ }
+
+
+
+ /**
+ * Deletes all files and subdirectories under "dir".
+ * @param dir Directory to be deleted
+ * @return boolean Returns "true" if all deletions were successful.
+ * If a deletion fails, the method stops attempting to
+ * delete and returns "false".
+ */
+ public static boolean deleteDir(File dir) {
+
+ if (dir.isDirectory()) {
+ String[] children = dir.list();
+ for (int i=0; i<children.length; i++) {
+ boolean success = deleteDir(new File(dir, children[i]));
+ if (!success) {
+ return false;
+ }
+ }
+ }
+
+ // The directory is now empty so now it can be smoked
+ return dir.delete();
+ }
+}