forgot to add new files from ZOOKEEPER-336

git-svn-id: https://svn.apache.org/repos/asf/hadoop/zookeeper/trunk@782883 13f79535-47bb-0310-9956-ffa450edef68

src/c/tests/TestClientRetry.cc

@@ -0,0 +1,496 @@
+/**
+ * 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.
+ */
+
+#include <cppunit/extensions/HelperMacros.h>
+#include "CppAssertHelper.h"
+
+#include <stdlib.h>
+#include <sys/select.h>
+
+#include "CollectionUtil.h"
+#include "ThreadingUtil.h"
+
+using namespace Util;
+
+#include "Vector.h"
+using namespace std;
+
+#include <cstring>
+#include <list>
+
+#include <zookeeper.h>
+
+#ifdef THREADED
+    static void yield(zhandle_t *zh, int i)
+    {
+        sleep(i);
+    }
+#else
+    static void yield(zhandle_t *zh, int seconds)
+    {
+        int fd;
+        int interest;
+        int events;
+        struct timeval tv;
+        int rc;
+        time_t expires = time(0) + seconds; 
+        time_t timeLeft = seconds;
+        fd_set rfds, wfds, efds;
+        FD_ZERO(&rfds);
+        FD_ZERO(&wfds);
+        FD_ZERO(&efds);
+
+        while(timeLeft >= 0) {
+            zookeeper_interest(zh, &fd, &interest, &tv);
+            if (fd != -1) {
+                if (interest&ZOOKEEPER_READ) {
+                    FD_SET(fd, &rfds);
+                } else {
+                    FD_CLR(fd, &rfds);
+                }
+                if (interest&ZOOKEEPER_WRITE) {
+                    FD_SET(fd, &wfds);
+                } else {
+                    FD_CLR(fd, &wfds);
+                }
+            } else {
+                fd = 0;
+            }
+            FD_SET(0, &rfds);
+            if (tv.tv_sec > timeLeft) {
+                tv.tv_sec = timeLeft;
+            }
+            rc = select(fd+1, &rfds, &wfds, &efds, &tv);
+            timeLeft = expires - time(0);
+            events = 0;
+            if (FD_ISSET(fd, &rfds)) {
+                events |= ZOOKEEPER_READ;
+            }
+            if (FD_ISSET(fd, &wfds)) {
+                events |= ZOOKEEPER_WRITE;
+            }
+            zookeeper_process(zh, events);
+        }
+    }
+#endif
+
+typedef struct evt {
+    string path;
+    int type;
+} evt_t;
+
+typedef struct watchCtx {
+private:
+    list<evt_t> events;
+public:
+    bool connected;
+    zhandle_t *zh;
+    Mutex mutex;
+
+    watchCtx() {
+        connected = false;
+        zh = 0;
+    }
+    ~watchCtx() {
+        if (zh) {
+            zookeeper_close(zh);
+            zh = 0;
+        }
+    }
+
+    evt_t getEvent() {
+        evt_t evt;
+        mutex.acquire();
+        CPPUNIT_ASSERT( events.size() > 0);
+        evt = events.front();
+        events.pop_front();
+        mutex.release();
+        return evt;
+    }
+
+    int countEvents() {
+        int count;
+        mutex.acquire();
+        count = events.size();
+        mutex.release();
+        return count;
+    }
+
+    void putEvent(evt_t evt) {
+        mutex.acquire();
+        events.push_back(evt);
+        mutex.release();
+    }
+
+    bool waitForConnected(zhandle_t *zh) {
+        time_t expires = time(0) + 10;
+        while(!connected && time(0) < expires) {
+            yield(zh, 1);
+        }
+        return connected;
+    }
+    bool waitForDisconnected(zhandle_t *zh) {
+        time_t expires = time(0) + 15;
+        while(connected && time(0) < expires) {
+            yield(zh, 1);
+        }
+        return !connected;
+    }
+} watchctx_t; 
+
+class Zookeeper_clientretry : public CPPUNIT_NS::TestFixture
+{
+    CPPUNIT_TEST_SUITE(Zookeeper_clientretry);
+#ifdef THREADED
+    CPPUNIT_TEST(testRetry);
+#endif
+    CPPUNIT_TEST_SUITE_END();
+
+    static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
+        watchctx_t *ctx = (watchctx_t*)v;
+
+        if (state == ZOO_CONNECTED_STATE) {
+            ctx->connected = true;
+        } else {
+            ctx->connected = false;
+        }
+        if (type != ZOO_SESSION_EVENT) {
+            evt_t evt;
+            evt.path = path;
+            evt.type = type;
+            ctx->putEvent(evt);
+        }
+    }
+
+    static const char hostPorts[];
+
+    const char *getHostPorts() {
+        return hostPorts;
+    }
+
+    zhandle_t *createClient(watchctx_t *ctx) {
+        zhandle_t *zk = zookeeper_init(hostPorts, watcher, 10000, 0,
+                                       ctx, 0);
+        ctx->zh = zk;
+        sleep(1);
+        return zk;
+    }
+    
+public:
+
+
+    void setUp()
+    {
+        char cmd[1024];
+        sprintf(cmd, "export ZKMAXCNXNS=1;%s startClean %s", ZKSERVER_CMD, getHostPorts());
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+    
+
+    void startServer() {
+        char cmd[1024];
+        sprintf(cmd, "export ZKMAXCNXNS=1;%s start %s", ZKSERVER_CMD, getHostPorts());
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+
+    void stopServer() {
+        tearDown();
+    }
+
+    void tearDown()
+    {
+        char cmd[1024];
+        sprintf(cmd, "%s stop %s", ZKSERVER_CMD, getHostPorts());
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+
+    bool waitForEvent(zhandle_t *zh, watchctx_t *ctx, int seconds) {
+        time_t expires = time(0) + seconds;
+        while(ctx->countEvents() == 0 && time(0) < expires) {
+            yield(zh, 1);
+        }
+        return ctx->countEvents() > 0;
+    }
+
+#define COUNT 100
+    
+    static zhandle_t *async_zk;
+
+    void testRetry()
+    {
+      watchctx_t ctx1, ctx2;
+      zhandle_t *zk1 = createClient(&ctx1);
+      CPPUNIT_ASSERT_EQUAL(true, ctx1.waitForConnected(zk1));
+      zhandle_t *zk2 = createClient(&ctx2);
+      zookeeper_close(zk1);
+      CPPUNIT_ASSERT_EQUAL(true, ctx2.waitForConnected(zk2));
+      ctx1.zh = 0;  
+    }
+};
+
+zhandle_t *Zookeeper_clientretry::async_zk;
+const char Zookeeper_clientretry::hostPorts[] = "127.0.0.1:22181";
+CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_clientretry);
+/**
+ * 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.
+ */
+
+#include <cppunit/extensions/HelperMacros.h>
+#include "CppAssertHelper.h"
+
+#include <stdlib.h>
+#include <sys/select.h>
+
+#include "CollectionUtil.h"
+#include "ThreadingUtil.h"
+
+using namespace Util;
+
+#include "Vector.h"
+using namespace std;
+
+#include <cstring>
+#include <list>
+
+#include <zookeeper.h>
+
+#ifdef THREADED
+    static void yield(zhandle_t *zh, int i)
+    {
+        sleep(i);
+    }
+#else
+    static void yield(zhandle_t *zh, int seconds)
+    {
+        int fd;
+        int interest;
+        int events;
+        struct timeval tv;
+        int rc;
+        time_t expires = time(0) + seconds; 
+        time_t timeLeft = seconds;
+        fd_set rfds, wfds, efds;
+        FD_ZERO(&rfds);
+        FD_ZERO(&wfds);
+        FD_ZERO(&efds);
+
+        while(timeLeft >= 0) {
+            zookeeper_interest(zh, &fd, &interest, &tv);
+            if (fd != -1) {
+                if (interest&ZOOKEEPER_READ) {
+                    FD_SET(fd, &rfds);
+                } else {
+                    FD_CLR(fd, &rfds);
+                }
+                if (interest&ZOOKEEPER_WRITE) {
+                    FD_SET(fd, &wfds);
+                } else {
+                    FD_CLR(fd, &wfds);
+                }
+            } else {
+                fd = 0;
+            }
+            FD_SET(0, &rfds);
+            if (tv.tv_sec > timeLeft) {
+                tv.tv_sec = timeLeft;
+            }
+            rc = select(fd+1, &rfds, &wfds, &efds, &tv);
+            timeLeft = expires - time(0);
+            events = 0;
+            if (FD_ISSET(fd, &rfds)) {
+                events |= ZOOKEEPER_READ;
+            }
+            if (FD_ISSET(fd, &wfds)) {
+                events |= ZOOKEEPER_WRITE;
+            }
+            zookeeper_process(zh, events);
+        }
+    }
+#endif
+
+typedef struct evt {
+    string path;
+    int type;
+} evt_t;
+
+typedef struct watchCtx {
+private:
+    list<evt_t> events;
+public:
+    bool connected;
+    zhandle_t *zh;
+    Mutex mutex;
+
+    watchCtx() {
+        connected = false;
+        zh = 0;
+    }
+    ~watchCtx() {
+        if (zh) {
+            zookeeper_close(zh);
+            zh = 0;
+        }
+    }
+
+    evt_t getEvent() {
+        evt_t evt;
+        mutex.acquire();
+        CPPUNIT_ASSERT( events.size() > 0);
+        evt = events.front();
+        events.pop_front();
+        mutex.release();
+        return evt;
+    }
+
+    int countEvents() {
+        int count;
+        mutex.acquire();
+        count = events.size();
+        mutex.release();
+        return count;
+    }
+
+    void putEvent(evt_t evt) {
+        mutex.acquire();
+        events.push_back(evt);
+        mutex.release();
+    }
+
+    bool waitForConnected(zhandle_t *zh) {
+        time_t expires = time(0) + 10;
+        while(!connected && time(0) < expires) {
+            yield(zh, 1);
+        }
+        return connected;
+    }
+    bool waitForDisconnected(zhandle_t *zh) {
+        time_t expires = time(0) + 15;
+        while(connected && time(0) < expires) {
+            yield(zh, 1);
+        }
+        return !connected;
+    }
+} watchctx_t; 
+
+class Zookeeper_clientretry : public CPPUNIT_NS::TestFixture
+{
+    CPPUNIT_TEST_SUITE(Zookeeper_clientretry);
+#ifdef THREADED
+    CPPUNIT_TEST(testRetry);
+#endif
+    CPPUNIT_TEST_SUITE_END();
+
+    static void watcher(zhandle_t *, int type, int state, const char *path,void*v){
+        watchctx_t *ctx = (watchctx_t*)v;
+
+        if (state == ZOO_CONNECTED_STATE) {
+            ctx->connected = true;
+        } else {
+            ctx->connected = false;
+        }
+        if (type != ZOO_SESSION_EVENT) {
+            evt_t evt;
+            evt.path = path;
+            evt.type = type;
+            ctx->putEvent(evt);
+        }
+    }
+
+    static const char hostPorts[];
+
+    const char *getHostPorts() {
+        return hostPorts;
+    }
+
+    zhandle_t *createClient(watchctx_t *ctx) {
+        zhandle_t *zk = zookeeper_init(hostPorts, watcher, 10000, 0,
+                                       ctx, 0);
+        ctx->zh = zk;
+        sleep(1);
+        return zk;
+    }
+    
+public:
+
+
+    void setUp()
+    {
+        char cmd[1024];
+        sprintf(cmd, "export ZKMAXCNXNS=1;%s startClean %s", ZKSERVER_CMD, getHostPorts());
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+    
+
+    void startServer() {
+        char cmd[1024];
+        sprintf(cmd, "export ZKMAXCNXNS=1;%s start %s", ZKSERVER_CMD, getHostPorts());
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+
+    void stopServer() {
+        tearDown();
+    }
+
+    void tearDown()
+    {
+        char cmd[1024];
+        sprintf(cmd, "%s stop %s", ZKSERVER_CMD, getHostPorts());
+        CPPUNIT_ASSERT(system(cmd) == 0);
+    }
+
+    bool waitForEvent(zhandle_t *zh, watchctx_t *ctx, int seconds) {
+        time_t expires = time(0) + seconds;
+        while(ctx->countEvents() == 0 && time(0) < expires) {
+            yield(zh, 1);
+        }
+        return ctx->countEvents() > 0;
+    }
+
+#define COUNT 100
+    
+    static zhandle_t *async_zk;
+
+    void testRetry()
+    {
+      watchctx_t ctx1, ctx2;
+      zhandle_t *zk1 = createClient(&ctx1);
+      CPPUNIT_ASSERT_EQUAL(true, ctx1.waitForConnected(zk1));
+      zhandle_t *zk2 = createClient(&ctx2);
+      zookeeper_close(zk1);
+      CPPUNIT_ASSERT_EQUAL(true, ctx2.waitForConnected(zk2));
+      ctx1.zh = 0;  
+    }
+};
+
+zhandle_t *Zookeeper_clientretry::async_zk;
+const char Zookeeper_clientretry::hostPorts[] = "127.0.0.1:22181";
+CPPUNIT_TEST_SUITE_REGISTRATION(Zookeeper_clientretry);

src/java/test/org/apache/zookeeper/test/ClientRetry.java

@@ -0,0 +1,60 @@
+/**
+ * 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.zookeeper.test;
+
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooKeeper.States;
+
+public class ClientRetry extends ClientBase{
+
+    public void setUp() throws Exception {
+        maxCnxns = 1;
+        super.setUp();     
+    }
+    /*
+     * This is a simple test - try to connect two clients to a server 
+     * accepting a maximum of one connection from each address. Check that 
+     * only one is accepted. Close that connection, and check that the other
+     * eventually connects. 
+     * 
+     * There is a possibility of a false positive here, as when zk2 is tested 
+     * for having connected it might not have been given enough time, and finish
+     * connecting after the test is done. Since the
+     * server doesn't tell the client why it hasn't connected, there's no 
+     * obvious way to detect the difference. 
+     */
+    public void testClientRetry() throws IOException, InterruptedException, TimeoutException{
+        CountdownWatcher cdw1 = new CountdownWatcher();
+        CountdownWatcher cdw2 = new CountdownWatcher();
+        ZooKeeper zk = new ZooKeeper(hostPort, 10000, cdw1);
+        cdw1.waitForConnected(CONNECTION_TIMEOUT);
+        ZooKeeper zk2 = new ZooKeeper(hostPort, 10000, cdw2);
+        States s1 = zk.getState();
+        States s2 = zk2.getState();
+        assertSame(s1,States.CONNECTED);
+        assertSame(s2,States.CONNECTING);
+        cdw1.reset();
+        zk.close();
+        cdw1.waitForDisconnected(CONNECTION_TIMEOUT);
+        cdw2.waitForConnected(CONNECTION_TIMEOUT);
+        assertSame(zk2.getState(),States.CONNECTED);
+    }
+}
+           

src/java/test/org/apache/zookeeper/test/MaxCnxnsTest.java

@@ -0,0 +1,103 @@
+/**
+ * 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.zookeeper.test;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+import org.apache.jute.BinaryOutputArchive;
+import org.apache.zookeeper.proto.ConnectRequest;
+
+public class MaxCnxnsTest extends ClientBase {
+
+    final private int numCnxns = 5;
+    
+    protected void setUp() throws Exception {
+        maxCnxns = numCnxns;
+        super.setUp();
+    }
+    
+    /**
+     * Verify the ability to limit the number of concurrent connections. 
+     * @throws IOException
+     * @throws InterruptedException
+     */
+    public void testMaxCnxns() throws IOException, InterruptedException{
+        SocketChannel[] sockets = new SocketChannel[numCnxns+5];
+        String split[] = hostPort.split(":");
+        String host = split[0];
+        int port = Integer.parseInt(split[1]);
+        int numConnected = 0;
+
+        /*
+         * For future unwary socket programmers: although connect 'blocks' it 
+         * does not require an accept on the server side to return. Therefore
+         * you can not assume that all the sockets are connected at the end of
+         * this for loop.
+         */
+        for (int i=0;i<sockets.length;++i) {            
+          SocketChannel sChannel = SocketChannel.open();
+          sChannel.connect(new InetSocketAddress(host,port));          
+          sockets[i] = sChannel;
+        }
+        // Construct a connection request
+        ConnectRequest conReq = new ConnectRequest(0, 0,
+                10000, 0, "password".getBytes());
+        ByteArrayOutputStream baos = new ByteArrayOutputStream();
+        BinaryOutputArchive boa = BinaryOutputArchive.getArchive(baos);
+        boa.writeInt(-1, "len");
+        conReq.serialize(boa, "connect");
+        baos.close();
+        ByteBuffer bb = ByteBuffer.wrap(baos.toByteArray());
+        bb.putInt(bb.capacity() - 4);
+        
+        /* Send a connect request. Any socket that has been closed (or at least
+         * not added to the cnxn list on the server) will not have any bytes to
+         * read and get an eof.
+         * 
+         *  The trick here was finding a call that caused the server to put
+         *  bytes in the input stream without closing the cnxn. None of
+         *  the four letter commands do that, so we actually try to create
+         *  a session which should send us something back, while maintaining
+         *  the connection.
+         */
+        for (int i=0;i<sockets.length;++i) {
+            try {
+                bb.rewind();
+                int eof = sockets[i].write(bb);
+                // If the socket times out, we count that as failed - 
+                // the server should respond within 10s
+                sockets[i].socket().setSoTimeout(10000);                
+                if (!sockets[i].socket().isClosed()){
+                    eof = sockets[i].socket().getInputStream().read(); 
+                    if (eof != -1) {
+                        numConnected++;
+                    }
+                }
+            }            
+            catch (IOException io) {
+                // "Connection reset by peer"
+            }
+        }
+        assertSame(numCnxns,numConnected);
+    }
+}