You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by br...@apache.org on 2009/06/09 07:23:27 UTC
svn commit: r782883 - in /hadoop/zookeeper/trunk/src:
c/tests/TestClientRetry.cc
java/test/org/apache/zookeeper/test/ClientRetry.java
java/test/org/apache/zookeeper/test/MaxCnxnsTest.java
Author: breed
Date: Tue Jun 9 05:23:26 2009
New Revision: 782883
URL: http://svn.apache.org/viewvc?rev=782883&view=rev
Log:
forgot to add new files from ZOOKEEPER-435
Added:
hadoop/zookeeper/trunk/src/c/tests/TestClientRetry.cc
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientRetry.java
hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/MaxCnxnsTest.java
Added: hadoop/zookeeper/trunk/src/c/tests/TestClientRetry.cc
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/c/tests/TestClientRetry.cc?rev=782883&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/c/tests/TestClientRetry.cc (added)
+++ hadoop/zookeeper/trunk/src/c/tests/TestClientRetry.cc Tue Jun 9 05:23:26 2009
@@ -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);
Added: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientRetry.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientRetry.java?rev=782883&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientRetry.java (added)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/ClientRetry.java Tue Jun 9 05:23:26 2009
@@ -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);
+ }
+}
+
Added: hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/MaxCnxnsTest.java
URL: http://svn.apache.org/viewvc/hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/MaxCnxnsTest.java?rev=782883&view=auto
==============================================================================
--- hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/MaxCnxnsTest.java (added)
+++ hadoop/zookeeper/trunk/src/java/test/org/apache/zookeeper/test/MaxCnxnsTest.java Tue Jun 9 05:23:26 2009
@@ -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);
+ }
+}