From hadoop-commits-returnderrickle@example.net Thu Nov 01 01:49:23 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 71077 invoked from network); 1 Nov 2007 01:49:22 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 01:49:22 -0000 Received: (qmail 34854 invoked by uid 500); 1 Nov 2007 01:49:10 -0000 Delivered-To: derrickle@example.net Received: (qmail 34818 invoked by uid 500); 1 Nov 2007 01:49:10 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 34809 invoked by uid 99); 1 Nov 2007 01:49:10 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 31 Oct 2007 18:49:09 -0700 X-ASF-Spam-Status: No, hits=-100.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 01:49:25 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 6DEB01A9832; Wed, 31 Oct 2007 18:48:47 -0700 (PDT) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r590875 - in /lucene/hadoop/trunk/src/contrib/hbase: ./ src/java/org/apache/hadoop/hbase/ src/java/org/apache/hadoop/hbase/util/ src/java/org/onelab/filter/ src/test/org/onelab/test/ Date: Thu, 01 Nov 2007 01:48:46 -0000 To: derrickle@example.net From: derrickle@example.net X-Mailer: svnmailer-1.0.8 Message-Id: <derrickle@example.net> X-Virus-Checked: Checked by ClamAV on apache.org Author: jimk Date: Wed Oct 31 18:48:46 2007 New Revision: 590875 URL: http://svn.apache.org/viewvc?rev=590875&view=rev Log: HADOOP-2126 Use Bob Jenkins' hash for bloom filters Added: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/util/JenkinsHash.java Modified: lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java lucene/hadoop/trunk/src/contrib/hbase/src/java/org/onelab/filter/HashFunction.java lucene/hadoop/trunk/src/contrib/hbase/src/test/org/onelab/test/TestFilter.java Modified: lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt?rev=590875&r1=590874&r2=590875&view=diff ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt (original) +++ lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt Wed Oct 31 18:48:46 2007 @@ -22,6 +22,7 @@ HADOOP-2401 Add convenience put method that takes writable (Johan Oskarsson via Stack) HADOOP-2088 Make hbase runnable in $HADOOP_HOME/build(/contrib/hbase) + HADOOP-2126 Use Bob Jenkins' hash for bloom filters Branch 0.15 (unreleased changes) Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java?rev=590875&r1=590874&r2=590875&view=diff ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java (original) +++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/HRegionInfo.java Wed Oct 31 18:48:46 2007 @@ -23,46 +23,24 @@ import java.io.DataOutput; import java.io.IOException; -import java.security.MessageDigest; -import java.security.NoSuchAlgorithmException; - import org.apache.hadoop.io.Text; import org.apache.hadoop.io.WritableComparable; +import org.apache.hadoop.hbase.util.JenkinsHash; + /** * HRegion information. * Contains HRegion id, start and end keys, a reference to this * HRegions' table descriptor, etc. */ public class HRegionInfo implements WritableComparable { - private static MessageDigest encoder = null; - - static { - try { - if (encoder == null) { - encoder = MessageDigest.getInstance("SHA"); - } - } catch (NoSuchAlgorithmException e) { - e.printStackTrace(); - } - } - /** * @param regionName * @return the encodedName */ public static String encodeRegionName(final Text regionName) { - byte[] bytes = null; - synchronized (encoder) { - encoder.update(regionName.getBytes(), 0, regionName.getLength()); - bytes = encoder.digest(); - encoder.reset(); - } - StringBuilder sb = new StringBuilder(); - for (int i = 0; i < bytes.length; i++) { - sb.append(bytes[i]); - } - return sb.toString(); + return String.valueOf( + JenkinsHash.hash(regionName.getBytes(), regionName.getLength(), 0)); } /** delimiter used between portions of a region name */ Added: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/util/JenkinsHash.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/util/JenkinsHash.java?rev=590875&view=auto ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/util/JenkinsHash.java (added) +++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/apache/hadoop/hbase/util/JenkinsHash.java Wed Oct 31 18:48:46 2007 @@ -0,0 +1,234 @@ +/** + * Copyright 2007 The Apache Software Foundation + * + * 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.hadoop.hbase.util; + +/** + * lookup3.c, by Bob Jenkins, May 2006, Public Domain. + * <a href="http://burtleburtle.net/bob/c/lookup3.c">lookup3.c</a> + * + * You can use this free for any purpose. It's in the public domain. + * It has no warranty. + * + * Produces 32-bit hash for hash table lookup. + */ +public class JenkinsHash { + private static long INT_MASK = 0x00000000ffffffffL; + private static long BYTE_MASK = 0x00000000000000ffL; + + private static long rot(long val, int pos) { + return Long.valueOf(Integer.rotateLeft( + Long.valueOf(val & INT_MASK).intValue(), pos)).longValue() & INT_MASK; + } + + /** + * Alternate form for hashing an entire byte array + * + * @param bytes + * @param initval + * @return hash value + */ + public static int hash(byte[] bytes, int initval) { + return hash(bytes, bytes.length, initval); + } + + /** + * taken from hashlittle() -- hash a variable-length key into a 32-bit value + * + * @param key the key (the unaligned variable-length array of bytes) + * @param nbytes number of bytes to include in hash + * @param initval can be any integer value + * @return a 32-bit value. Every bit of the key affects every bit of the + * return value. Two keys differing by one or two bits will have totally + * different hash values. + * + * The best hash table sizes are powers of 2. There is no need to do mod a + * prime (mod is sooo slow!). If you need less than 32 bits, use a bitmask. + * For example, if you need only 10 bits, do h = (h & hashmask(10)); + * In which case, the hash table should have hashsize(10) elements. + * + * If you are hashing n strings byte[][] k, do it like this: + * for (int i = 0, h = 0; i < n; ++i) h = hash( k[i], h); + * + * By Bob Jenkins, 2006. derrickle@example.net. You may use this + * code any way you wish, private, educational, or commercial. It's free. + * + * Use for hash table lookup, or anything where one collision in 2^^32 is + * acceptable. Do NOT use for cryptographic purposes. + */ + public static int hash(byte[] key, int nbytes, int initval) { + int length = nbytes; + long a, b, c; // We use longs because we don't have unsigned ints + a = b = c = (0x00000000deadbeefL + length + initval) & INT_MASK; + int offset = 0; + for (; length > 12; offset += 12, length -= 12) { + a = (a + (key[offset + 0] & BYTE_MASK)) & INT_MASK; + a = (a + (((key[offset + 1] & BYTE_MASK) << 8) & INT_MASK)) & INT_MASK; + a = (a + (((key[offset + 2] & BYTE_MASK) << 16) & INT_MASK)) & INT_MASK; + a = (a + (((key[offset + 3] & BYTE_MASK) << 24) & INT_MASK)) & INT_MASK; + b = (b + (key[offset + 4] & BYTE_MASK)) & INT_MASK; + b = (b + (((key[offset + 5] & BYTE_MASK) << 8) & INT_MASK)) & INT_MASK; + b = (b + (((key[offset + 6] & BYTE_MASK) << 16) & INT_MASK)) & INT_MASK; + b = (b + (((key[offset + 7] & BYTE_MASK) << 24) & INT_MASK)) & INT_MASK; + c = (c + (key[offset + 8] & BYTE_MASK)) & INT_MASK; + c = (c + (((key[offset + 9] & BYTE_MASK) << 8) & INT_MASK)) & INT_MASK; + c = (c + (((key[offset + 10] & BYTE_MASK) << 16) & INT_MASK)) & INT_MASK; + c = (c + (((key[offset + 11] & BYTE_MASK) << 24) & INT_MASK)) & INT_MASK; + + /* + * mix -- mix 3 32-bit values reversibly. + * This is reversible, so any information in (a,b,c) before mix() is + * still in (a,b,c) after mix(). + * + * If four pairs of (a,b,c) inputs are run through mix(), or through + * mix() in reverse, there are at least 32 bits of the output that + * are sometimes the same for one pair and different for another pair. + * + * This was tested for: + * - pairs that differed by one bit, by two bits, in any combination + * of top bits of (a,b,c), or in any combination of bottom bits of + * (a,b,c). + * - "differ" is defined as +, -, ^, or ~^. For + and -, I transformed + * the output delta to a Gray code (a^(a>>1)) so a string of 1's (as + * is commonly produced by subtraction) look like a single 1-bit + * difference. + * - the base values were pseudorandom, all zero but one bit set, or + * all zero plus a counter that derrickle@example.net. + * + * Some k values for my "a-=c; a^=rot(c,k); c+=b;" arrangement that + * satisfy this are + * 4 6 8 16 19 4 + * +1-720-534-9480 + * 14 9 3 7 17 3 + * Well, "+1-720-534-9480" didn't quite get 32 bits diffing for + * "differ" defined as + with a one-bit base and a two-bit delta. I + * used http://burtleburtle.net/bob/hash/avalanche.html to choose + * the operations, constants, and arrangements of the variables. + * + * This does not achieve avalanche. There are input bits of (a,b,c) + * that fail to affect some output bits of (a,b,c), especially of a. + * The most thoroughly mixed value is c, but it doesn't really even + * achieve avalanche in c. + * + * This allows some parallelism. Read-after-writes are good at doubling + * the number of bits affected, so the goal of mixing pulls in the + * opposite direction as the goal of parallelism. I did what I could. + * Rotates seem to cost as much as shifts on every machine I could lay + * my hands on, and rotates are much kinder to the top and bottom bits, + * so I used rotates. + * + * #define mix(a,b,c) \ + * { \ + * a -= c; a ^= rot(c, 4); c += b; \ + * b -= a; b ^= rot(a, 6); a += c; \ + * c -= b; c ^= rot(b, 8); b += a; \ + * a -= c; a ^= rot(c,16); c += b; \ + * b -= a; b ^= rot(a,19); a += c; \ + * c -= b; c ^= rot(b, 4); b += a; \ + * } + * + * mix(a,b,c); + */ + a = (a - c) & INT_MASK; a ^= rot(c, 4); c = (c + b) & INT_MASK; + b = (b - a) & INT_MASK; b ^= rot(a, 6); a = (a + c) & INT_MASK; + c = (c - b) & INT_MASK; c ^= rot(b, 8); b = (b + a) & INT_MASK; + a = (a - c) & INT_MASK; a ^= rot(c,16); c = (c + b) & INT_MASK; + b = (b - a) & INT_MASK; b ^= rot(a,19); a = (a + c) & INT_MASK; + c = (c - b) & INT_MASK; c ^= rot(b, 4); b = (b + a) & INT_MASK; + } + + //-------------------------------- last block: affect all 32 bits of (c) + switch (length) { // all the case statements fall through + case 12: + c = (c + (((key[offset + 11] & BYTE_MASK) << 24) & INT_MASK)) & INT_MASK; + case 11: + c = (c + (((key[offset + 10] & BYTE_MASK) << 16) & INT_MASK)) & INT_MASK; + case 10: + c = (c + (((key[offset + 9] & BYTE_MASK) << 8) & INT_MASK)) & INT_MASK; + case 9: + c = (c + (key[offset + 8] & BYTE_MASK)) & INT_MASK; + case 8: + b = (b + (((key[offset + 7] & BYTE_MASK) << 24) & INT_MASK)) & INT_MASK; + case 7: + b = (b + (((key[offset + 6] & BYTE_MASK) << 16) & INT_MASK)) & INT_MASK; + case 6: + b = (b + (((key[offset + 5] & BYTE_MASK) << 8) & INT_MASK)) & INT_MASK; + case 5: + b = (b + (key[offset + 4] & BYTE_MASK)) & INT_MASK; + case 4: + a = (a + (((key[offset + 3] & BYTE_MASK) << 24) & INT_MASK)) & INT_MASK; + case 3: + a = (a + (((key[offset + 2] & BYTE_MASK) << 16) & INT_MASK)) & INT_MASK; + case 2: + a = (a + (((key[offset + 1] & BYTE_MASK) << 8) & INT_MASK)) & INT_MASK; + case 1: + a = (a + (key[offset + 0] & BYTE_MASK)) & INT_MASK; + break; + case 0: + return Long.valueOf(c & INT_MASK).intValue(); + } + /* + * final -- final mixing of 3 32-bit values (a,b,c) into c + * + * Pairs of (a,b,c) values differing in only a few bits will usually + * produce values of c that look totally different. This was tested for + * - pairs that differed by one bit, by two bits, in any combination + * of top bits of (a,b,c), or in any combination of bottom bits of + * (a,b,c). + * + * - "differ" is defined as +, -, ^, or ~^. For + and -, I transformed + * the output delta to a Gray code (a^(a>>1)) so a string of 1's (as + * is commonly produced by subtraction) look like a single 1-bit + * difference. + * + * - the base values were pseudorandom, all zero but one bit set, or + * all zero plus a counter that derrickle@example.net. + * + * These constants passed: + * 14 11 25 16 4 14 24 + * 12 14 25 16 4 14 24 + * and these came close: + * 4 8 15 26 3 22 24 + * 10 8 15 26 3 22 24 + * +1-720-534-9480 + * + * #define final(a,b,c) \ + * { + * c ^= b; c -= rot(b,14); \ + * a ^= c; a -= rot(c,11); \ + * b ^= a; b -= rot(a,25); \ + * c ^= b; c -= rot(b,16); \ + * a ^= c; a -= rot(c,4); \ + * b ^= a; b -= rot(a,14); \ + * c ^= b; c -= rot(b,24); \ + * } + * + */ + c ^= b; c = (c - rot(b,14)) & INT_MASK; + a ^= c; a = (a - rot(c,11)) & INT_MASK; + b ^= a; b = (b - rot(a,25)) & INT_MASK; + c ^= b; c = (c - rot(b,16)) & INT_MASK; + a ^= c; a = (a - rot(c,4)) & INT_MASK; + b ^= a; b = (b - rot(a,14)) & INT_MASK; + c ^= b; c = (c - rot(b,24)) & INT_MASK; + + return Long.valueOf(c & INT_MASK).intValue(); + } +} Modified: lucene/hadoop/trunk/src/contrib/hbase/src/java/org/onelab/filter/HashFunction.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/java/org/onelab/filter/HashFunction.java?rev=590875&r1=590874&r2=590875&view=diff ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/src/java/org/onelab/filter/HashFunction.java (original) +++ lucene/hadoop/trunk/src/contrib/hbase/src/java/org/onelab/filter/HashFunction.java Wed Oct 31 18:48:46 2007 @@ -49,7 +49,7 @@ */ package org.onelab.filter; -import java.security.*; +import org.apache.hadoop.hbase.util.JenkinsHash; /** * Implements a hash object that returns a certain number of hashed values. @@ -66,9 +66,6 @@ * @see <a href="http://www.itl.nist.gov/fipspubs/fip180-1.htm">SHA-1 algorithm</a> */ public final class HashFunction{ - /** The SHA-1 algorithm. */ - private MessageDigest sha; - /** The number of hashed values. */ private int nbHash; @@ -83,13 +80,6 @@ * @param nbHash The number of resulting hashed values. */ public HashFunction(int maxValue, int nbHash) { - try { - sha = MessageDigest.getInstance("SHA-1"); - - } catch(NoSuchAlgorithmException e) { - throw new AssertionError(e); - } - if(maxValue <= 0) { throw new IllegalArgumentException("maxValue must be > 0"); } @@ -102,9 +92,8 @@ this.nbHash = nbHash; }//end constructor - /** Clears <i>this</i> hash function. */ + /** Clears <i>this</i> hash function. A NOOP */ public void clear(){ - sha.reset(); }//end clear() /** @@ -121,19 +110,9 @@ if(b.length == 0) { throw new IllegalArgumentException("key length must be > 0"); } - sha.update(b); - byte[] digestBytes = sha.digest(); int[] result = new int[nbHash]; - int nbBytePerInt = digestBytes.length/nbHash; - int offset = 0; - for(int i = 0; i < nbHash; i++){ - int val = 0; - for(int j = offset; j < offset + nbBytePerInt; j++) { - val |= - (digestBytes[offset] & 0xff) << ((nbBytePerInt - 1 - (j - offset)) * 8); - } - result[i] = Math.abs(val) % maxValue; - offset += nbBytePerInt; + for (int i = 0, initval = 0; i < nbHash; i++) { + result[i] = Math.abs(JenkinsHash.hash(b, initval)) % maxValue; } return result; }//end hash() Modified: lucene/hadoop/trunk/src/contrib/hbase/src/test/org/onelab/test/TestFilter.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/test/org/onelab/test/TestFilter.java?rev=590875&r1=590874&r2=590875&view=diff ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/src/test/org/onelab/test/TestFilter.java (original) +++ lucene/hadoop/trunk/src/contrib/hbase/src/test/org/onelab/test/TestFilter.java Wed Oct 31 18:48:46 2007 @@ -75,8 +75,8 @@ bf.add(k3); assertTrue(bf.membershipTest(key)); assertFalse(bf.membershipTest(new StringKey("graknyl"))); - assertTrue(bf.membershipTest(new StringKey("xyzzy"))); // False positive - assertTrue(bf.membershipTest(new StringKey("abcd"))); // False positive + assertFalse(bf.membershipTest(new StringKey("xyzzy"))); + assertFalse(bf.membershipTest(new StringKey("abcd"))); } /** Test a CountingBloomFilter @@ -92,8 +92,8 @@ bf.add(k3); assertTrue(bf.membershipTest(key)); assertFalse(bf.membershipTest(new StringKey("graknyl"))); - assertTrue(bf.membershipTest(new StringKey("xyzzy"))); // False positive - assertTrue(bf.membershipTest(new StringKey("abcd"))); // False positive + assertFalse(bf.membershipTest(new StringKey("xyzzy"))); + assertFalse(bf.membershipTest(new StringKey("abcd"))); } /** Test a DynamicBloomFilter @@ -110,6 +110,6 @@ assertTrue(bf.membershipTest(key)); assertFalse(bf.membershipTest(new StringKey("graknyl"))); assertFalse(bf.membershipTest(new StringKey("xyzzy"))); - assertTrue(bf.membershipTest(new StringKey("abcd"))); // False positive + assertFalse(bf.membershipTest(new StringKey("abcd"))); } }//end class From hadoop-commits-returnderrickle@example.net Thu Nov 01 06:25:15 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 79979 invoked from network); 1 Nov 2007 06:25:15 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 06:25:15 -0000 Received: (qmail 74197 invoked by uid 500); 1 Nov 2007 06:25:02 -0000 Delivered-To: derrickle@example.net Received: (qmail 74079 invoked by uid 500); 1 Nov 2007 06:25:02 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 74070 invoked by uid 99); 1 Nov 2007 06:25:02 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Wed, 31 Oct 2007 23:25:02 -0700 X-ASF-Spam-Status: No, hits=-100.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 06:25:28 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 3D46159A07 for <derrickle@example.net>; Thu, 1 Nov 2007 06:24:51 +0000 (GMT) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 8bit From: Apache Wiki <derrickle@example.net> To: derrickle@example.net Date: Thu, 01 Nov 2007 06:24:51 -0000 Message-ID: <derrickle@example.net> Subject: [Lucene-hadoop Wiki] Trivial Update of "Hbase/Matrix" by udanax X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Lucene-hadoop Wiki" for change notification. The following page has been changed by udanax: http://wiki.apache.org/lucene-hadoop/Hbase/Matrix The comment on the change is: add reference ------------------------------------------------------------------------------ * ScaLAPACK, a library of high-performance linear algebra routines for distributed-memory message-passing MIMD computers * Scheduling algorithms for parallel Gaussian elimination withcommunication costs, Amoura, A.K.; Bampis, E.; Konig, J.-C. * High performance numerical libraries in Java, Bjørn-Ove Heimsund + * Google's PageRank and Beyond: The Science of Search Engine Rankigs, Amy N. Langville; Carl D. Meyer From hadoop-commits-returnderrickle@example.net Thu Nov 01 08:50:57 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 30000 invoked from network); 1 Nov 2007 08:50:57 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 08:50:57 -0000 Received: (qmail 74213 invoked by uid 500); 1 Nov 2007 08:50:45 -0000 Delivered-To: derrickle@example.net Received: (qmail 74180 invoked by uid 500); 1 Nov 2007 08:50:45 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 74171 invoked by uid 99); 1 Nov 2007 08:50:45 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 01:50:45 -0700 X-ASF-Spam-Status: No, hits=-100.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 08:51:01 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id DF31C59A07 for <derrickle@example.net>; Thu, 1 Nov 2007 08:50:23 +0000 (GMT) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 8bit From: Apache Wiki <derrickle@example.net> To: derrickle@example.net Date: Thu, 01 Nov 2007 08:50:23 -0000 Message-ID: <derrickle@example.net> Subject: [Lucene-hadoop Wiki] Trivial Update of "Hbase/Matrix" by udanax X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Lucene-hadoop Wiki" for change notification. The following page has been changed by udanax: http://wiki.apache.org/lucene-hadoop/Hbase/Matrix ------------------------------------------------------------------------------ * [http://icl.cs.utk.edu/iclprojects/pages/files/sans/yelick-bebop.pdf Automatic Performance Tuning of Sparse Matrix Kernels], August 7, 2002 * Parallel Conjugate Gradients Assignment, a parallel implementation of the conjugate gradient algorithm * ScaLAPACK, a library of high-performance linear algebra routines for distributed-memory message-passing MIMD computers - * Scheduling algorithms for parallel Gaussian elimination withcommunication costs, Amoura, A.K.; Bampis, E.; Konig, J.-C. + * Scheduling algorithms for parallel Gaussian elimination with communication costs, Amoura, A.K.; Bampis, E.; Konig, J.-C. * High performance numerical libraries in Java, Bjørn-Ove Heimsund * Google's PageRank and Beyond: The Science of Search Engine Rankigs, Amy N. Langville; Carl D. Meyer From hadoop-commits-returnderrickle@example.net Thu Nov 01 09:02:06 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 33150 invoked from network); 1 Nov 2007 09:02:06 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 09:02:06 -0000 Received: (qmail 84620 invoked by uid 500); 1 Nov 2007 09:01:54 -0000 Delivered-To: derrickle@example.net Received: (qmail 84580 invoked by uid 500); 1 Nov 2007 09:01:54 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 84571 invoked by uid 99); 1 Nov 2007 09:01:54 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 02:01:54 -0700 X-ASF-Spam-Status: No, hits=-100.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 09:02:05 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 86EC75A250 for <derrickle@example.net>; Thu, 1 Nov 2007 09:01:45 +0000 (GMT) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: Apache Wiki <derrickle@example.net> To: derrickle@example.net Date: Thu, 01 Nov 2007 09:01:45 -0000 Message-ID: <derrickle@example.net> Subject: [Lucene-hadoop Wiki] Update of "Hbase/HbaseFeatureRequests" by rschuil X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Lucene-hadoop Wiki" for change notification. The following page has been changed by rschuil: http://wiki.apache.org/lucene-hadoop/Hbase/HbaseFeatureRequests ------------------------------------------------------------------------------ * Clients should be able to open an !OutputStream for putting a value into a table. Right now the only interface for insertion is via a byte[], which severely limits the flexibility of the system. Added ''2007-10-08 by stuhood'' * Tell us more Stu why you need this feature? (Because you want to write BLOBs of multiple Gigabytes into an hbase cell?) + * Add ability to read a cell's timestamp. This can be used for sorting columns or removing old columns. + From hadoop-commits-returnderrickle@example.net Thu Nov 01 15:17:59 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 55298 invoked from network); 1 Nov 2007 15:17:59 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 15:17:59 -0000 Received: (qmail 50100 invoked by uid 500); 1 Nov 2007 15:01:53 -0000 Delivered-To: derrickle@example.net Received: (qmail 50074 invoked by uid 500); 1 Nov 2007 15:01:53 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 50063 invoked by uid 99); 1 Nov 2007 15:01:53 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 08:01:53 -0700 X-ASF-Spam-Status: No, hits=-98.0 required=10.0 tests=ALL_TRUSTED,URIBL_BLACK X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 15:02:04 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 81FE61A9832; Thu, 1 Nov 2007 08:01:44 -0700 (PDT) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r591039 - in /lucene/hadoop/trunk/src/contrib/hbase: CHANGES.txt src/test/org/apache/hadoop/hbase/MultiRegionTable.java Date: Thu, 01 Nov 2007 15:01:44 -0000 To: derrickle@example.net From: derrickle@example.net X-Mailer: svnmailer-1.0.8 Message-Id: <derrickle@example.net> X-Virus-Checked: Checked by ClamAV on apache.org Author: stack Date: Thu Nov 1 08:01:43 2007 New Revision: 591039 URL: http://svn.apache.org/viewvc?rev=591039&view=rev Log: HADOOP-2083 TestTableIndex failed in patch build #970 and #956 Modified: lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MultiRegionTable.java Modified: lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt?rev=591039&r1=591038&r2=591039&view=diff ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt (original) +++ lucene/hadoop/trunk/src/contrib/hbase/CHANGES.txt Thu Nov 1 08:01:43 2007 @@ -17,6 +17,7 @@ HADOOP-2056 A table with row keys containing colon fails to split regions HADOOP-2079 HADOOP-2056 Fix generated HLog, HRegion names HADOOP-2124 Use of `hostname` does not work on Cygwin in some cases + HADOOP-2083 TestTableIndex failed in #970 and #956 IMPROVEMENTS HADOOP-2401 Add convenience put method that takes writable Modified: lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MultiRegionTable.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MultiRegionTable.java?rev=591039&r1=591038&r2=591039&view=diff ============================================================================== --- lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MultiRegionTable.java (original) +++ lucene/hadoop/trunk/src/contrib/hbase/src/test/org/apache/hadoop/hbase/MultiRegionTable.java Thu Nov 1 08:01:43 2007 @@ -75,19 +75,27 @@ HTable meta = new HTable(conf, HConstants.META_TABLE_NAME); int count = count(meta, tableName); HTable t = new HTable(conf, new Text(tableName)); + // We created the table. Get the parent region here now. One will + // have been created though nought in it. + HRegionInfo parent = + t.getRegionLocation(HConstants.EMPTY_START_ROW).getRegionInfo(); + LOG.info("Parent region " + parent.toString()); + // Now add content. addContent(new HTableIncommon(t), columnName); LOG.info("Finished content loading"); // All is running in the one JVM so I should be able to get the single - // region instance and bring on a split. - // Presumption is that there is only one regionserver. + // region instance and bring on a split. Presumption is that there is only + // one regionserver. Of not, the split may already have happened by the + // time we got here. If so, then the region found when we go searching + // with EMPTY_START_ROW will be one of the unsplittable daughters. HRegionInfo hri = null; HRegion r = null; for (int i = 0; i < 30; i++) { hri = t.getRegionLocation(HConstants.EMPTY_START_ROW).getRegionInfo(); LOG.info("Region location: " + hri); r = cluster.getRegionThreads().get(0).getRegionServer(). - onlineRegions.get(hri.getRegionName()); + onlineRegions.get(hri.getRegionName()); if (r != null) { break; } @@ -121,9 +129,8 @@ // Get info on the parent from the meta table. Pass in 'hri'. Its the // region we have been dealing with up to this. Its the parent of the // region split. - Map<Text, byte []> data = getSplitParentInfo(meta, hri); - HRegionInfo parent = - Writables.getHRegionInfoOrNull(data.get(HConstants.COL_REGIONINFO)); + Map<Text, byte []> data = getSplitParentInfo(meta, parent); + parent = Writables.getHRegionInfoOrNull(data.get(HConstants.COL_REGIONINFO)); LOG.info("Found parent region: " + parent); assertTrue(parent.isOffline()); assertTrue(parent.isSplit()); From hadoop-commits-returnderrickle@example.net Thu Nov 01 17:14:21 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 1205 invoked from network); 1 Nov 2007 17:14:21 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 17:14:21 -0000 Received: (qmail 87355 invoked by uid 500); 1 Nov 2007 17:12:53 -0000 Delivered-To: derrickle@example.net Received: (qmail 87335 invoked by uid 500); 1 Nov 2007 17:12:53 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 87322 invoked by uid 99); 1 Nov 2007 17:12:53 -0000 Received: from nike.apache.org (HELO nike.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 10:12:53 -0700 X-ASF-Spam-Status: No, hits=-100.0 required=10.0 tests=ALL_TRUSTED X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eos.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 17:13:21 +0000 Received: from eos.apache.org (localhost [83.161.120.155]) by eos.apache.org (Postfix) with ESMTP id 14F8E59A07 for <derrickle@example.net>; Thu, 1 Nov 2007 17:12:43 +0000 (GMT) Content-Type: text/plain; charset="us-ascii" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit From: Apache Wiki <derrickle@example.net> To: derrickle@example.net Date: Thu, 01 Nov 2007 17:12:42 -0000 Message-ID: <derrickle@example.net> Subject: [Lucene-hadoop Wiki] Trivial Update of "Hbase/HbaseFeatureRequests" by JimKellerman X-Virus-Checked: Checked by ClamAV on apache.org Dear Wiki user, You have subscribed to a wiki page or wiki category on "Lucene-hadoop Wiki" for change notification. The following page has been changed by JimKellerman: http://wiki.apache.org/lucene-hadoop/Hbase/HbaseFeatureRequests ------------------------------------------------------------------------------ * Add ability to read a cell's timestamp. This can be used for sorting columns or removing old columns. + This is possible with scanners as next() returns the row key and the time stamp in the H!StoreKey. Are you asking for a similar mechanism for get? + From hadoop-commits-returnderrickle@example.net Thu Nov 01 18:09:04 2007 Return-Path: <hadoop-commits-returnderrickle@example.net> Delivered-To: derrickle@example.net Received: (qmail 17327 invoked from network); 1 Nov 2007 18:09:04 -0000 Received: from hermes.apache.org (HELO mail.apache.org) (83.161.120.155) by minotaur.apache.org with SMTP; 1 Nov 2007 18:09:04 -0000 Received: (qmail 66411 invoked by uid 500); 1 Nov 2007 18:08:52 -0000 Delivered-To: derrickle@example.net Received: (qmail 66378 invoked by uid 500); 1 Nov 2007 18:08:52 -0000 Mailing-List: contact derrickle@example.net; run by ezmlm Precedence: bulk List-Help: <mailto:derrickle@example.net> List-Unsubscribe: <mailto:derrickle@example.net> List-Post: <mailto:derrickle@example.net> List-Id: <hadoop-commits.lucene.apache.org> Reply-To: derrickle@example.net Delivered-To: mailing list derrickle@example.net Received: (qmail 66369 invoked by uid 99); 1 Nov 2007 18:08:52 -0000 Received: from athena.apache.org (HELO athena.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 11:08:52 -0700 X-ASF-Spam-Status: No, hits=-97.3 required=10.0 tests=ALL_TRUSTED,FRT_LEVITRA,URIBL_BLACK X-Spam-Check-By: apache.org Received: from [83.161.120.155] (HELO eris.apache.org) (83.161.120.155) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 01 Nov 2007 18:09:02 +0000 Received: by eris.apache.org (Postfix, from userid 65534) id 604681A983A; Thu, 1 Nov 2007 11:08:42 -0700 (PDT) Content-Type: text/plain; charset="utf-8" MIME-Version: 1.0 Content-Transfer-Encoding: 7bit Subject: svn commit: r591103 - in /lucene/hadoop/trunk: ./ src/java/org/apache/hadoop/dfs/ src/test/org/apache/hadoop/dfs/ Date: Thu, 01 Nov 2007 18:08:41 -0000 To: derrickle@example.net From: derrickle@example.net X-Mailer: svnmailer-1.0.8 Message-Id: <derrickle@example.net> X-Virus-Checked: Checked by ClamAV on apache.org Author: dhruba Date: Thu Nov 1 11:08:40 2007 New Revision: 591103 URL: http://svn.apache.org/viewvc?rev=591103&view=rev Log: HADOOP-1912. Datanode has two new commands COPY and REPLACE. These are needed for supporting data rebalance. (Hairong Kuang via dhruba) Added: lucene/hadoop/trunk/src/test/org/apache/hadoop/dfs/TestBlockReplacement.java (with props) Modified: lucene/hadoop/trunk/CHANGES.txt lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java lucene/hadoop/trunk/src/test/org/apache/hadoop/dfs/DFSTestUtil.java Modified: lucene/hadoop/trunk/CHANGES.txt URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/CHANGES.txt?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/CHANGES.txt (original) +++ lucene/hadoop/trunk/CHANGES.txt Thu Nov 1 11:08:40 2007 @@ -35,6 +35,9 @@ HADOOP-1210. Log counters in job history. (Owen O'Malley via ddas) + HADOOP-1912. Datanode has two new commands COPY and REPLACE. These are + needed for supporting data rebalance. (Hairong Kuang via dhruba) + OPTIMIZATIONS HADOOP-1898. Release the lock protecting the last time of the last stack Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java (original) +++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DataNode.java Thu Nov 1 11:08:40 2007 @@ -21,6 +21,7 @@ import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.IOUtils; +import org.apache.hadoop.io.Text; import org.apache.hadoop.ipc.*; import org.apache.hadoop.conf.*; import org.apache.hadoop.metrics.MetricsUtil; @@ -36,8 +37,10 @@ import java.io.*; import java.net.*; import java.util.*; +import java.util.concurrent.Semaphore; import java.security.NoSuchAlgorithmException; import java.security.SecureRandom; + import org.apache.hadoop.metrics.MetricsContext; import org.apache.hadoop.metrics.MetricsRecord; import org.apache.hadoop.metrics.Updater; @@ -109,7 +112,9 @@ DatanodeRegistration dnRegistration = null; private String networkLoc; volatile boolean shouldRun = true; - LinkedList<Block> receivedBlockList = new LinkedList<Block>(); + private LinkedList<Block> receivedBlockList = new LinkedList<Block>(); + private LinkedList<String> delHints = new LinkedList<String>(); + final private static String EMPTY_DEL_HINT = ""; int xmitsInProgress = 0; Daemon dataXceiveServer = null; long blockReportInterval; @@ -124,6 +129,13 @@ private Thread dataNodeThread = null; String machineName; int defaultBytesPerChecksum = 512; + + // The following three fields are to support balancing + final private static long BALANCE_BANDWIDTH = 1024L*1024; // 1MB/s + final private static short MAX_BALANCING_THREADS = 5; + private Semaphore balancingSem = new Semaphore(MAX_BALANCING_THREADS); + private Throttler balancingThrottler = new Throttler(BALANCE_BANDWIDTH); + private static class DataNodeMetrics implements Updater { private final MetricsRecord metricsRecord; private int bytesWritten = 0; @@ -531,19 +543,28 @@ // check if there are newly received blocks Block [] blockArray=null; + String [] delHintArray=null; synchronized(receivedBlockList) { - if (receivedBlockList.size() > 0) { - // - // Send newly-received blockids to namenode - // - blockArray = receivedBlockList.toArray(new Block[receivedBlockList.size()]); + synchronized(delHints) { + int numBlocks = receivedBlockList.size(); + if (receivedBlockList.size() > 0) { + assert(numBlocks==delHints.size()); + // + // Send newly-received blockids to namenode + // + blockArray = receivedBlockList.toArray(new Block[numBlocks]); + delHintArray = delHints.toArray(new String[numBlocks]); + } } } if (blockArray != null) { - namenode.blockReceived(dnRegistration, blockArray); + namenode.blockReceived(dnRegistration, blockArray, delHintArray); synchronized (receivedBlockList) { - for(Block b: blockArray) { - receivedBlockList.remove(b); + synchronized (delHints) { + for(int i=0; i<blockArray.length; i++) { + receivedBlockList.remove(blockArray[i]); + delHints.remove(delHintArray[i]); + } } } } @@ -725,15 +746,16 @@ * till namenode is informed before responding with success to the * client? For now we don't. */ - private void notifyNamenodeReceivedBlock(Block block) { + private void notifyNamenodeReceivedBlock(Block block, String delHint) { synchronized (receivedBlockList) { - receivedBlockList.add(block); - receivedBlockList.notifyAll(); + synchronized (delHints) { + receivedBlockList.add(block); + delHints.add(delHint); + receivedBlockList.notifyAll(); + } } } - - /** * Server used for receiving/sending a block of data. * This is created to listen for requests from clients or @@ -752,8 +774,6 @@ try { while (shouldRun) { Socket s = ss.accept(); - //s.setSoTimeout(READ_TIMEOUT); - xceiverCount.incr(); new Daemon(new DataXceiver(s)).start(); } ss.close(); @@ -806,13 +826,18 @@ case OP_READ_METADATA: readMetadata( in ); break; + case OP_REPLACE_BLOCK: // for balancing purpose; send to a destination + replaceBlock(in); + break; + case OP_COPY_BLOCK: // for balancing purpose; send to a proxy source + copyBlock(in); + break; default: throw new IOException("Unknown opcode " + op + "in data stream"); } } catch (Throwable t) { LOG.error("DataXceiver: " + StringUtils.stringifyException(t)); } finally { - xceiverCount.decr(); LOG.debug("Number of active connections is: "+xceiverCount); IOUtils.closeStream(in); IOUtils.closeSocket(s); @@ -825,6 +850,7 @@ * @throws IOException */ private void readBlock(DataInputStream in) throws IOException { + xceiverCount.incr(); // // Read in the header // @@ -864,6 +890,7 @@ StringUtils.stringifyException(ioe) ); throw ioe; } finally { + xceiverCount.decr(); IOUtils.closeStream(out); IOUtils.closeStream(blockSender); } @@ -876,6 +903,7 @@ * @throws IOException */ private void writeBlock(DataInputStream in) throws IOException { + xceiverCount.incr(); // // Read in the header // @@ -935,10 +963,11 @@ // receive the block and mirror to the next target String mirrorAddr = (mirrorSock == null) ? null : mirrorNode; + blockReceiver.receiveBlock(mirrorOut, mirrorAddr, null); // notify name node - notifyNamenodeReceivedBlock(block); + notifyNamenodeReceivedBlock(block, EMPTY_DEL_HINT); String msg = "Received block " + block + " from " + s.getInetAddress(); @@ -971,6 +1000,8 @@ IOUtils.closeStream(mirrorOut); IOUtils.closeSocket(mirrorSock); IOUtils.closeStream(blockReceiver); + // decrement counter + xceiverCount.decr(); } } @@ -979,7 +1010,8 @@ * @param in */ void readMetadata(DataInputStream in) throws IOException { - + xceiverCount.incr(); + Block block = new Block( in.readLong(), 0 ); InputStream checksumIn = null; DataOutputStream out = null; @@ -1007,21 +1039,190 @@ //last DATA_CHUNK out.writeInt(0); } finally { + xceiverCount.decr(); IOUtils.closeStream(checksumIn); } } + + /** + * Read a block from the disk and then sends it to a destination + * + * @param in + * The stream to read from + * @throws IOException + */ + private void copyBlock(DataInputStream in) throws IOException { + // Read in the header + long blockId = in.readLong(); // read block id + Block block = new Block(blockId, 0); + String source = Text.readString(in); // read del hint + DatanodeInfo target = new DatanodeInfo(); // read target + target.readFields(in); + + Socket targetSock = null; + short opStatus = OP_STATUS_SUCCESS; + BlockSender blockSender = null; + DataOutputStream targetOut = null; + try { + balancingSem.acquireUninterruptibly(); + + // check if the block exists or not + blockSender = new BlockSender(block, 0, -1, false, false); + + // get the output stream to the target + InetSocketAddress targetAddr = createSocketAddr(target.getName()); + targetSock = new Socket(); + targetSock.connect(targetAddr, READ_TIMEOUT); + targetSock.setSoTimeout(READ_TIMEOUT); + + targetOut = new DataOutputStream(new BufferedOutputStream( + targetSock.getOutputStream(), BUFFER_SIZE)); + + /* send request to the target */ + // fist write header info + targetOut.writeShort(DATA_TRANFER_VERSION); // transfer version + targetOut.writeByte(OP_REPLACE_BLOCK); // op code + targetOut.writeLong(block.getBlockId()); // block id + Text.writeString( targetOut, source); // del hint + + // then send data + long read = blockSender.sendBlock(targetOut, balancingThrottler); + + myMetrics.readBytes((int) read); + myMetrics.readBlocks(1); + + // check the response from target + receiveResponse(targetSock); + + LOG.info("Copied block " + block + " to " + targetAddr); + } catch (IOException ioe) { + opStatus = OP_STATUS_ERROR; + LOG.warn("Got exception while serving " + block + " to " + + target.getName() + ": " + StringUtils.stringifyException(ioe)); + throw ioe; + } finally { + /* send response to the requester */ + try { + sendResponse(s, opStatus); + } catch (IOException replyE) { + LOG.warn("Error writing the response back to "+ + s.getRemoteSocketAddress() + "\n" + + StringUtils.stringifyException(replyE) ); + } + IOUtils.closeStream(targetOut); + IOUtils.closeStream(blockSender); + balancingSem.release(); + } + } + + /** + * Receive a block and write it to disk, it then notifies the namenode to + * remove the copy from the source + * + * @param in + * The stream to read from + * @throws IOException + */ + private void replaceBlock(DataInputStream in) throws IOException { + balancingSem.acquireUninterruptibly(); + + /* read header */ + Block block = new Block(in.readLong(), 0); // block id & len + String sourceID = Text.readString(in); + + short opStatus = OP_STATUS_SUCCESS; + BlockReceiver blockReceiver = null; + try { + // open a block receiver and check if the block does not exist + blockReceiver = new BlockReceiver( + block, in, s.getRemoteSocketAddress().toString()); + + // receive a block + blockReceiver.receiveBlock(null, null, balancingThrottler); + + // notify name node + notifyNamenodeReceivedBlock(block, sourceID); + + LOG.info("Received block " + block + + " from " + s.getRemoteSocketAddress()); + } catch (IOException ioe) { + opStatus = OP_STATUS_ERROR; + throw ioe; + } finally { + // send response back + try { + sendResponse(s, opStatus); + } catch (IOException ioe) { + LOG.warn("Error writing reply back to " + s.getRemoteSocketAddress()); + } + IOUtils.closeStream(blockReceiver); + balancingSem.release(); + } + } } - /* An interface to throttle the block transfers */ - private interface Throttler { + /** a class to throttle the block transfers + * This class is thread safe. It can be shared by multiple threads. + * The parameter bandwidthPerSec specifies the total bandwidth shared by threads. + */ + static class Throttler { + private long period; // period over which bw is imposed + private long periodExtension; // Max period over which bw accumulates. + private long bytesPerPeriod; // total number of bytes can be sent in each period + private long curPeriodStart; // current period starting time + private long curReserve; // remaining bytes can be sent in the period + private long bytesAlreadyUsed; + + /** Constructor */ + Throttler(long bandwidthPerSec) { + this(1000, bandwidthPerSec); // by default throttling period is 1s + } + + /** Constructor */ + Throttler(long period, long bandwidthPerSec) { + this.curPeriodStart = System.currentTimeMillis(); + this.period = period; + this.curReserve = this.bytesPerPeriod = bandwidthPerSec*period/1000; + this.periodExtension = period*3; + } + /** Given the numOfBytes sent/received since last time throttle was called, - * make the current thread sleep if I/O rate is too fast + * make the current thread sleep if I/O rate is too fast * compared to the given bandwidth - * - * @param numOfBytes + * + * @param numOfBytes * number of bytes sent/received since last time throttle was called - */ - void throttle(int numOfBytes); + */ + public synchronized void throttle(long numOfBytes) { + if ( numOfBytes <= 0 ) { + return; + } + + curReserve -= numOfBytes; + bytesAlreadyUsed += numOfBytes; + + while (curReserve <= 0) { + long now = System.currentTimeMillis(); + long curPeriodEnd = curPeriodStart + period; + + if ( now < curPeriodEnd ) { + // Wait for next period so that curReserve can be increased. + try { + wait( curPeriodEnd - now ); + } catch (InterruptedException ignored) {} + } else if ( now < (curPeriodStart + periodExtension)) { + curPeriodStart = curPeriodEnd; + curReserve += bytesPerPeriod; + } else { + // discard the prev period. Throttler might not have + // been used for a long time. + curPeriodStart = now; + curReserve = bytesPerPeriod - bytesAlreadyUsed; + } + } + + bytesAlreadyUsed -= numOfBytes; + } } private class BlockSender implements java.io.Closeable { @@ -1175,7 +1376,7 @@ out.write(buf, 0, len + checksumSize); if (throttler != null) { // rebalancing so throttle - throttler.throttle(len + checksumSize); + throttler.throttle(len + checksumSize + 4); } return len; @@ -1354,7 +1555,7 @@ } if (throttler != null) { // throttle I/O - throttler.throttle(len + checksumSize); + throttler.throttle(len + checksumSize + 4); } } Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java (original) +++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DatanodeProtocol.java Thu Nov 1 11:08:40 2007 @@ -31,9 +31,9 @@ **********************************************************************/ interface DatanodeProtocol extends VersionedProtocol { /* - * 9: heartbeat sends also the data node used space; + * 10: blockReceived also sends hints for deletion */ - public static final long versionID = 9L; + public static final long versionID = 10L; // error code final static int NOTIFY = 0; @@ -89,12 +89,15 @@ /** * blockReceived() allows the DataNode to tell the NameNode about - * recently-received block data. For example, whenever client code + * recently-received block data, with a hint for pereferred replica + * to be deleted when there is any excessive blocks. + * For example, whenever client code * writes a new Block here, or another DataNode copies a Block to * this DataNode, it will call blockReceived(). */ public void blockReceived(DatanodeRegistration registration, - Block blocks[]) throws IOException; + Block blocks[], + String[] delHints) throws IOException; /** * errorReport() tells the NameNode about something that has gone Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java (original) +++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java Thu Nov 1 11:08:40 2007 @@ -85,6 +85,8 @@ public static final byte OP_WRITE_BLOCK = (byte) 80; public static final byte OP_READ_BLOCK = (byte) 81; public static final byte OP_READ_METADATA = (byte) 82; + public static final byte OP_REPLACE_BLOCK = (byte) 83; + public static final byte OP_COPY_BLOCK = (byte) 84; public static final int OP_STATUS_SUCCESS = 0; public static final int OP_STATUS_ERROR = 1; @@ -97,13 +99,21 @@ * This should change when serialization of DatanodeInfo, not just * when protocol changes. It is not very obvious. */ - /* Version 6: - * 0 marks the end of a block not an EMPTY_CHUNK - * OP_READ_BLOCK: return OP_STATUS_ERROR if received an invalid block id - * return OP_STATUS_ERROR if received an invalid length - * OP_WRITE_BLOCK: return OP_STATUS_ERROR if illegal bytesPerChecksum + /* Version 7: + * Add two operations to data node + * OP_COPY_BLOCK: + * The command is for sending to a proxy source for the balancing purpose + * The datanode then sends OP_REPLACE_BLOCK request to the destination + * OP_COPY_BLOCK BlockID(long) SourceID (UTF8) Destination (DatanodeInfo) + * return OP_STATUS_ERROR if any error occurs; OP_STATUS_SUCCESS otherwise + * OP_REPLACE_BLOCK: + * the command is for sending to a destination for the balancing purpose + * The datanode then writes the block to disk and notifies namenode of this + * received block together with a deletion hint: sourceID + * OP_REPLACE_BLOCK BlockID(long) SourceID(UTF8) Block_Data_With_Crc + * return OP_STATUS_ERROR if any error occurs; OP_STATUS_SUCCESS otherwise */ - public static final int DATA_TRANFER_VERSION = 6; //Should it be 1? + public static final int DATA_TRANFER_VERSION = 7; // Return codes for file create public static final int OPERATION_FAILED = 0; Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java (original) +++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java Thu Nov 1 11:08:40 2007 @@ -606,24 +606,22 @@ * return the length of the added block; 0 if the block is not added */ private long addBlock(Block block, List<BlockWithLocations> results) { - int numNodes = blocksMap.numNodes(block); - String[] machineSet = new String[numNodes]; - if (numNodes > 0) { - numNodes = 0; - for(Iterator<DatanodeDescriptor> it = - blocksMap.nodeIterator(block); it.hasNext();) { - String storageID = it.next().getStorageID(); - // filter invalidate replicas - Collection<Block> blocks = recentInvalidateSets.get(storageID); - if(blocks==null || !blocks.contains(block)) { - machineSet[numNodes++] = storageID; - } + ArrayList<String> machineSet = + new ArrayList<String>(blocksMap.numNodes(block)); + for(Iterator<DatanodeDescriptor> it = + blocksMap.nodeIterator(block); it.hasNext();) { + String storageID = it.next().getStorageID(); + // filter invalidate replicas + Collection<Block> blocks = recentInvalidateSets.get(storageID); + if(blocks==null || !blocks.contains(block)) { + machineSet.add(storageID); } } - if(numNodes == 0) { + if(machineSet.size() == 0) { return 0; } else { - results.add(new BlockWithLocations(block, machineSet)); + results.add(new BlockWithLocations(block, + machineSet.toArray(new String[machineSet.size()]))); return block.getNumBytes(); } } @@ -770,7 +768,7 @@ LOG.info("Reducing replication for file " + src + ". New replication is " + replication); for(int idx = 0; idx < fileBlocks.length; idx++) - proccessOverReplicatedBlock(fileBlocks[idx], replication); + proccessOverReplicatedBlock(fileBlocks[idx], replication, null, null); } return true; } @@ -2219,7 +2217,7 @@ removeStoredBlock(b, node); } for (Block b : toAdd) { - addStoredBlock(b, node); + addStoredBlock(b, node, null); } // @@ -2262,7 +2260,9 @@ * needed replications if this takes care of the problem. * @return the block that is stored in blockMap. */ - synchronized Block addStoredBlock(Block block, DatanodeDescriptor node) { + synchronized Block addStoredBlock(Block block, + DatanodeDescriptor node, + DatanodeDescriptor delNodeHint) { INodeFile fileINode = blocksMap.getINode(block); int replication = (fileINode != null) ? fileINode.getReplication() : @@ -2333,7 +2333,7 @@ updateNeededReplications(block, curReplicaDelta, 0); } if (numCurrentReplica > fileReplication) { - proccessOverReplicatedBlock(block, fileReplication); + proccessOverReplicatedBlock(block, fileReplication, node, delNodeHint); } return block; } @@ -2343,7 +2343,11 @@ * If there are any extras, call chooseExcessReplicates() to * mark them in the excessReplicateMap. */ - private void proccessOverReplicatedBlock(Block block, short replication) { + private void proccessOverReplicatedBlock(Block block, short replication, + DatanodeDescriptor addedNode, DatanodeDescriptor delNodeHint) { + if(addedNode == delNodeHint) { + delNodeHint = null; + } Collection<DatanodeDescriptor> nonExcess = new ArrayList<DatanodeDescriptor>(); for (Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(block); it.hasNext();) { @@ -2355,7 +2359,8 @@ } } } - chooseExcessReplicates(nonExcess, block, replication); + chooseExcessReplicates(nonExcess, block, replication, + addedNode, delNodeHint); } /** @@ -2373,7 +2378,9 @@ * then pick a node with least free space */ void chooseExcessReplicates(Collection<DatanodeDescriptor> nonExcess, - Block b, short replication) { + Block b, short replication, + DatanodeDescriptor addedNode, + DatanodeDescriptor delNodeHint) { // first form a rack to datanodes map and HashMap<String, ArrayList<DatanodeDescriptor>> rackMap = new HashMap<String, ArrayList<DatanodeDescriptor>>(); @@ -2405,22 +2412,29 @@ } } - // pick one node with least space from priSet if it is not empty + // pick one node to delete that favors the delete hint + // otherwise pick one with least space from priSet if it is not empty // otherwise one node with least space from remains while (nonExcess.size() - replication > 0) { DatanodeInfo cur = null; long minSpace = Long.MAX_VALUE; - Iterator<DatanodeDescriptor> iter = - priSet.isEmpty() ? remains.iterator() : priSet.iterator(); - while( iter.hasNext() ) { - DatanodeDescriptor node = iter.next(); - long free = node.getRemaining(); - - if (minSpace > free) { - minSpace = free; - cur = node; - } + // check if we can del delNodeHint + if( delNodeHint !=null && (priSet.contains(delNodeHint) || + (addedNode != null && !priSet.contains(addedNode))) ) { + cur = delNodeHint; + } else { // regular excessive replica removal + Iterator<DatanodeDescriptor> iter = + priSet.isEmpty() ? remains.iterator() : priSet.iterator(); + while( iter.hasNext() ) { + DatanodeDescriptor node = iter.next(); + long free = node.getRemaining(); + + if (minSpace > free) { + minSpace = free; + cur = node; + } + } } // adjust rackmap, priSet, and remains @@ -2430,14 +2444,13 @@ if(datanodes.isEmpty()) { rackMap.remove(rack); } - if (priSet.isEmpty()) { - remains.remove(cur); - } else { - priSet.remove(cur); + if( priSet.remove(cur) ) { if (datanodes.size() == 1) { priSet.remove(datanodes.get(0)); remains.add(datanodes.get(0)); } + } else { + remains.remove(cur); } nonExcess.remove(cur); @@ -2515,7 +2528,8 @@ * The given node is reporting that it received a certain block. */ public synchronized void blockReceived(DatanodeID nodeID, - Block block + Block block, + String delHint ) throws IOException { DatanodeDescriptor node = getDatanode(nodeID); if (node == null) { @@ -2538,10 +2552,22 @@ throw new DisallowedDatanodeException(node); } + // get the deletion hint node + DatanodeDescriptor delHintNode = null; + if(delHint!=null && delHint.length()!=0) { + delHintNode = datanodeMap.get(delHint); + if(delHintNode == null) { + NameNode.stateChangeLog.warn("BLOCK* NameSystem.blockReceived: " + + block.getBlockName() + + " is expected to be removed from an unrecorded node " + + delHint); + } + } + // // Modify the blocks->datanode map and node's map. // - addStoredBlock(block, node); + addStoredBlock(block, node, delHintNode ); pendingReplications.remove(block); } Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java (original) +++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java Thu Nov 1 11:08:40 2007 @@ -596,12 +596,13 @@ } public void blockReceived(DatanodeRegistration nodeReg, - Block blocks[]) throws IOException { + Block blocks[], + String delHints[]) throws IOException { verifyRequest(nodeReg); stateChangeLog.debug("*BLOCK* NameNode.blockReceived: " +"from "+nodeReg.getName()+" "+blocks.length+" blocks."); for (int i = 0; i < blocks.length; i++) { - namesystem.blockReceived(nodeReg, blocks[i]); + namesystem.blockReceived(nodeReg, blocks[i], delHints[i]); } } Modified: lucene/hadoop/trunk/src/test/org/apache/hadoop/dfs/DFSTestUtil.java URL: http://svn.apache.org/viewvc/lucene/hadoop/trunk/src/test/org/apache/hadoop/dfs/DFSTestUtil.java?rev=591103&r1=591102&r2=591103&view=diff ============================================================================== --- lucene/hadoop/trunk/src/test/org/apache/hadoop/dfs/DFSTestUtil.java (original) +++ lucene/hadoop/trunk/src/test/org/apache/hadoop/dfs/DFSTestUtil.java Thu Nov 1 11:08:40 2007 @@ -109,19 +109,29 @@ Path root = new Path(topdir); for (int idx = 0; idx < nFiles; idx++) { - Path fPath = new Path(root, files[idx].getName()); - if (!fs.mkdirs(fPath.getParent())) { - throw new IOException("Mkdirs failed to create " + - fPath.getParent().toString()); - } - FSDataOutputStream out = fs.create(fPath, replicationFactor); - byte[] toWrite = new byte[files[idx].getSize()]; - Random rb = new Random(files[idx].getSeed()); - rb.nextBytes(toWrite); - out.write(toWrite); - out.close(); - toWrite = null; + createFile(fs, new Path(root, files[idx].getName()), files[idx].getSize(), + replicationFactor, files[idx].getSeed()); + } + } + + static void createFile(FileSystem fs, Path fileName, long fileLen, + short replFactor, long seed) throws IOException { + if (!fs.mkdirs(fileName.getParent())) { + throw new IOException("Mkdirs failed to create " + + fileName.getParent().toString()); + } + FSDataOutputStream out = fs.create(fileName, replFactor); + byte[] toWrite = new byte[1024]; + Random rb = new Random(seed); + long bytesToWrite = fileLen; + while (bytesToWrite>0) { + rb.nextBytes(toWrite); + int bytesToWriteNext = (1024<bytesToWrite)?1024:(int)bytesToWrite; + + out.write(toWrite, 0, bytesToWriteNext); + bytesToWrite -= bytesToWriteNext; } + out.close(); } /** check if the files have been copied correctly. */ @@ -168,33 +178,31 @@ Path root = new Path(topdir); /** wait for the replication factor to settle down */ - while (true) { - boolean good = true; - for (int idx = 0; idx < nFiles; idx++) { - Path fPath = new Path(root, files[idx].getName()); - String locs[][] = fs.getFileCacheHints(fPath,
{'source': 'Apache Software Foundation Public Mail Archives'}