zookeeper-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From iv...@apache.org
Subject svn commit: r1239068 - in /zookeeper/bookkeeper/trunk: ./ bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/ bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/ bookkeeper-...
Date Wed, 01 Feb 2012 11:06:33 GMT
Author: ivank
Date: Wed Feb  1 11:06:32 2012
New Revision: 1239068

URL: http://svn.apache.org/viewvc?rev=1239068&view=rev
Log:
BOOKKEEPER-23: Timeout requests (ivank)

Added:
    zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestReadTimeout.java
Modified:
    zookeeper/bookkeeper/trunk/CHANGES.txt
    zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
    zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
    zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/NIOServerFactory.java
    zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
    zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BaseTestCase.java

Modified: zookeeper/bookkeeper/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/CHANGES.txt?rev=1239068&r1=1239067&r2=1239068&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/CHANGES.txt (original)
+++ zookeeper/bookkeeper/trunk/CHANGES.txt Wed Feb  1 11:06:32 2012
@@ -26,6 +26,8 @@ Trunk (unreleased changes)
 
         BOOKKEEPER-153: Ledger can't be opened or closed due to zero-length metadata (Sijie
Guo via ivank)
 
+        BOOKKEEPER-23: Timeout requests (ivank)
+
       hedwig-server/
       
         BOOKKEEPER-140: Hub server doesn't subscribe remote region correctly when a region
is down. (Sijie Gou via ivank)

Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java?rev=1239068&r1=1239067&r2=1239068&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
(original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Bookie.java
Wed Feb  1 11:06:32 2012
@@ -245,6 +245,7 @@ public class Bookie extends Thread {
 
     public Bookie(ServerConfiguration conf) 
             throws IOException, KeeperException, InterruptedException {
+        super("Bookie-" + conf.getBookiePort());
         this.conf = conf;
         this.journalDirectory = conf.getJournalDir();
         this.ledgerDirectories = conf.getLedgerDirs();

Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java?rev=1239068&r1=1239067&r2=1239068&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
(original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/conf/ClientConfiguration.java
Wed Feb  1 11:06:32 2012
@@ -38,6 +38,7 @@ public class ClientConfiguration extends
 
     // NIO Parameters
     protected final static String CLIENT_TCP_NODELAY = "clientTcpNoDelay";
+    protected final static String READ_TIMEOUT = "readTimeout";
 
     /**
      * Construct a default client-side configuration
@@ -213,4 +214,25 @@ public class ClientConfiguration extends
         return this;
     }
 
+    /**
+     * Get the socket read timeout. This is the number of
+     * seconds we wait without hearing a response from a bookie
+     * before we consider it failed.
+     *
+     * @return the current read timeout in seconds
+     */
+    public int getReadTimeout() {
+        return getInt(READ_TIMEOUT, 5);
+    }
+
+    /**
+     * Set the socket read timeout.
+     * @see #getReadTimeout()
+     * @param timeout The new read timeout in seconds
+     * @return client configuration
+     */
+    public ClientConfiguration setReadTimeout(int timeout) {
+        setProperty(READ_TIMEOUT, Integer.toString(timeout));
+        return this;
+    }
 }

Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/NIOServerFactory.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/NIOServerFactory.java?rev=1239068&r1=1239067&r2=1239068&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/NIOServerFactory.java
(original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/NIOServerFactory.java
Wed Feb  1 11:06:32 2012
@@ -75,7 +75,7 @@ public class NIOServerFactory extends Th
     ServerConfiguration conf;
 
     public NIOServerFactory(ServerConfiguration conf, PacketProcessor processor) throws IOException
{
-        super("NIOServerFactory");
+        super("NIOServerFactory-" + conf.getBookiePort());
         setDaemon(true);
         this.processor = processor;
         this.conf = conf;

Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java?rev=1239068&r1=1239067&r2=1239068&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
(original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/main/java/org/apache/bookkeeper/proto/PerChannelBookieClient.java
Wed Feb  1 11:06:32 2012
@@ -36,6 +36,8 @@ import org.apache.bookkeeper.util.Ordere
 import org.apache.bookkeeper.util.SafeRunnable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.jboss.netty.util.Timer;
+import org.jboss.netty.util.HashedWheelTimer;
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
@@ -56,6 +58,8 @@ import org.jboss.netty.channel.socket.Cl
 import org.jboss.netty.handler.codec.frame.CorruptedFrameException;
 import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
 import org.jboss.netty.handler.codec.frame.TooLongFrameException;
+import org.jboss.netty.handler.timeout.ReadTimeoutHandler;
+import org.jboss.netty.handler.timeout.ReadTimeoutException;
 
 /**
  * This class manages all details of connection to a particular bookie. It also
@@ -76,6 +80,7 @@ public class PerChannelBookieClient exte
     AtomicLong totalBytesOutstanding;
     ClientSocketChannelFactory channelFactory;
     OrderedSafeExecutor executor;
+    private Timer readTimeoutTimer;
 
     ConcurrentHashMap<CompletionKey, AddCompletion> addCompletions = new ConcurrentHashMap<CompletionKey,
AddCompletion>();
     ConcurrentHashMap<CompletionKey, ReadCompletion> readCompletions = new ConcurrentHashMap<CompletionKey,
ReadCompletion>();
@@ -107,6 +112,7 @@ public class PerChannelBookieClient exte
         this.totalBytesOutstanding = totalBytesOutstanding;
         this.channelFactory = channelFactory;
         this.state = ConnectionState.DISCONNECTED;
+        this.readTimeoutTimer = new HashedWheelTimer();
     }
 
     synchronized private void connect() {
@@ -375,6 +381,9 @@ public class PerChannelBookieClient exte
     @Override
     public ChannelPipeline getPipeline() throws Exception {
         ChannelPipeline pipeline = Channels.pipeline();
+
+        pipeline.addLast("readTimeout", new ReadTimeoutHandler(readTimeoutTimer, 
+                                                               conf.getReadTimeout()));
         pipeline.addLast("lengthbasedframedecoder", new LengthFieldBasedFrameDecoder(MAX_FRAME_LENGTH,
0, 4, 0, 4));
         pipeline.addLast("mainhandler", this);
         return pipeline;
@@ -388,6 +397,7 @@ public class PerChannelBookieClient exte
         LOG.info("Disconnected from bookie: " + addr);
         errorOutOutstandingEntries();
         channel.close();
+        readTimeoutTimer.stop();
 
         state = ConnectionState.DISCONNECTED;
 
@@ -407,6 +417,11 @@ public class PerChannelBookieClient exte
                       + e.getChannel().getRemoteAddress());
             return;
         }
+        if (t instanceof ReadTimeoutException) {
+            ctx.getChannel().disconnect();
+            return;
+        }
+ 
         if (t instanceof IOException) {
             // these are thrown when a bookie fails, logging them just pollutes
             // the logs (the failure is logged from the listeners on the write

Added: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestReadTimeout.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestReadTimeout.java?rev=1239068&view=auto
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestReadTimeout.java
(added)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/client/TestReadTimeout.java
Wed Feb  1 11:06:32 2012
@@ -0,0 +1,101 @@
+package org.apache.bookkeeper.client;
+
+/*
+ *
+ * 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.
+ *
+ */
+
+import org.junit.*;
+import java.net.InetSocketAddress;
+import java.util.Enumeration;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
+
+import org.apache.bookkeeper.client.LedgerHandle;
+import org.apache.bookkeeper.client.LedgerEntry;
+import org.apache.bookkeeper.client.BookKeeper;
+import org.apache.bookkeeper.client.BookKeeperAdmin;
+import org.apache.bookkeeper.client.BKException;
+import org.apache.bookkeeper.client.BookKeeper.DigestType;
+import org.apache.bookkeeper.test.BaseTestCase;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This unit test tests ledger fencing;
+ *
+ */
+public class TestReadTimeout extends BaseTestCase {
+    static Logger LOG = LoggerFactory.getLogger(TestReadTimeout.class);
+
+    DigestType digestType;
+
+    public TestReadTimeout(DigestType digestType) {
+        super(10);
+        this.digestType = digestType;
+    }
+
+    @Test
+    public void testReadTimeout() throws Exception {
+        final AtomicBoolean completed = new AtomicBoolean(false);
+
+        LedgerHandle writelh = bkc.createLedger(3,3,digestType, "testPasswd".getBytes());
+        String tmp = "Foobar";
+        
+        final int numEntries = 10;
+        for (int i = 0; i < numEntries; i++) {
+            writelh.addEntry(tmp.getBytes());
+        }
+        
+        Set<InetSocketAddress> beforeSet = new HashSet<InetSocketAddress>();
+        for (InetSocketAddress addr : writelh.getLedgerMetadata().getEnsemble(numEntries))
{
+            beforeSet.add(addr);
+        }
+
+        final InetSocketAddress bookieToSleep 
+            = writelh.getLedgerMetadata().getEnsemble(numEntries).get(0);
+        int sleeptime = baseClientConf.getReadTimeout()*3;
+        CountDownLatch latch = new CountDownLatch(1);
+        sleepBookie(bookieToSleep, sleeptime, latch);
+        latch.await();
+
+        writelh.asyncAddEntry(tmp.getBytes(), 
+                new AddCallback() {
+                    public void addComplete(int rc, LedgerHandle lh, 
+                                            long entryId, Object ctx) {
+                        completed.set(true);
+                    }
+                }, null);
+        Thread.sleep((baseClientConf.getReadTimeout()*2)*1000);
+        Assert.assertTrue("Write request did not finish", completed.get());
+
+        Set<InetSocketAddress> afterSet = new HashSet<InetSocketAddress>();
+        for (InetSocketAddress addr : writelh.getLedgerMetadata().getEnsemble(numEntries+1))
{
+            afterSet.add(addr);
+        }
+        beforeSet.removeAll(afterSet);
+        Assert.assertTrue("Bookie set should not match", beforeSet.size() != 0);
+    }
+}

Modified: zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BaseTestCase.java
URL: http://svn.apache.org/viewvc/zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BaseTestCase.java?rev=1239068&r1=1239067&r2=1239068&view=diff
==============================================================================
--- zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BaseTestCase.java
(original)
+++ zookeeper/bookkeeper/trunk/bookkeeper-server/src/test/java/org/apache/bookkeeper/test/BaseTestCase.java
Wed Feb  1 11:06:32 2012
@@ -29,6 +29,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 
 import org.apache.bookkeeper.client.BookKeeperTestClient;
 import org.apache.bookkeeper.client.BookKeeper.DigestType;
@@ -166,6 +167,33 @@ public abstract class BaseTestCase exten
         }
     }
 
+    public void sleepBookie(InetSocketAddress addr, final int seconds,
+                            final CountDownLatch l)
+            throws InterruptedException, IOException {
+        final String name = "Bookie-" + addr.getPort();
+        Thread[] allthreads = new Thread[Thread.activeCount()];
+        Thread.enumerate(allthreads);
+        for (final Thread t : allthreads) {
+            if (t.getName().equals(name)) {
+                Thread sleeper = new Thread() {
+                    public void run() {
+                        try {
+                            t.suspend();
+                            l.countDown();
+                            Thread.sleep(seconds*1000);
+                            t.resume();
+                        } catch (Exception e) {
+                            LOG.error("Error suspending thread", e);
+                        }
+                    }
+                };
+                sleeper.start();
+                return;
+            }
+        }
+        throw new IOException("Bookie thread not found");
+    }
+
     /**
      * Restart bookie servers
      *



Mime
View raw message