ZOOKEEPER-1933. Windows release build of zk client cannot connect to zk server (Orion Hodson via fpj, phunt)

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

CHANGES.txt

@@ -744,6 +744,9 @@ BUGFIXES:
   ZOOKEEPER-1974. winvs2008 jenkins job failing with "unresolved
   external symbol" (flavio via phunt)
 
+  ZOOKEEPER-1933. Windows release build of zk client cannot connect to
+  zk server (Orion Hodson via fpj, phunt)
+
 IMPROVEMENTS:
 
   ZOOKEEPER-1170. Fix compiler (eclipse) warnings: unused imports,

src/c/src/mt_adaptor.c

@@ -369,6 +369,7 @@ void *do_io(void *v)
     fds[0].fd=adaptor_threads->self_pipe[0];
     fds[0].events=POLLIN;
     while(!zh->close_requested) {
+        zh->io_count++;
         struct timeval tv;
         int fd;
         int interest;
@@ -396,45 +397,56 @@ void *do_io(void *v)
             while(read(adaptor_threads->self_pipe[0],b,sizeof(b))==sizeof(b)){}
         }        
 #else
-    fd_set rfds, wfds, efds;
+    fd_set rfds, wfds;
     struct adaptor_threads *adaptor_threads = zh->adaptor_priv;
     api_prolog(zh);
     notify_thread_ready(zh);
     LOG_DEBUG(LOGCALLBACK(zh), "started IO thread");
-    FD_ZERO(&rfds);   FD_ZERO(&wfds);    FD_ZERO(&efds);
+    
     while(!zh->close_requested) {      
         struct timeval tv;
         SOCKET fd;
-        SOCKET maxfd=adaptor_threads->self_pipe[0];
-        int interest;        
+        int interest;
         int rc;
-               
-       zookeeper_interest(zh, &fd, &interest, &tv);
-       if (fd != -1) {
-           if (interest&ZOOKEEPER_READ) {
+
+        zookeeper_interest(zh, &fd, &interest, &tv);
+
+        // FD_ZERO is cheap on Win32, it just sets count of elements to zero.
+        // It needs to be done to ensure no stale entries.
+        FD_ZERO(&rfds);
+        FD_ZERO(&wfds);
+
+        if (fd != -1) {
+            if (interest&ZOOKEEPER_READ) {
                 FD_SET(fd, &rfds);
-            } else {
-                FD_CLR(fd, &rfds);
             }
-           if (interest&ZOOKEEPER_WRITE) {
+
+            if (interest&ZOOKEEPER_WRITE) {
                 FD_SET(fd, &wfds);
-            } else {
-                FD_CLR(fd, &wfds);
-            }                  
+            }
         }
-       FD_SET( adaptor_threads->self_pipe[0] ,&rfds );        
-       rc = select((int)maxfd, &rfds, &wfds, &efds, &tv);
-       if (fd != -1) 
-       {
-           interest = (FD_ISSET(fd, &rfds))? ZOOKEEPER_READ:0;
-           interest|= (FD_ISSET(fd, &wfds))? ZOOKEEPER_WRITE:0;
+
+        // Always interested in self_pipe.
+        FD_SET(adaptor_threads->self_pipe[0], &rfds);
+
+        rc = select(/* unused */0, &rfds, &wfds, NULL, &tv);
+        if (rc > 0) {
+            interest=(FD_ISSET(fd, &rfds))? ZOOKEEPER_READ: 0;
+            interest|=(FD_ISSET(fd, &wfds))? ZOOKEEPER_WRITE: 0;
+
+            if (FD_ISSET(adaptor_threads->self_pipe[0], &rfds)){
+                // flush the pipe/socket
+                char b[128];
+                while(recv(adaptor_threads->self_pipe[0],b,sizeof(b), 0)==sizeof(b)){}
+            }
         }
-               
-       if (FD_ISSET(adaptor_threads->self_pipe[0], &rfds)){
-            // flush the pipe/socket
-            char b[128];
-           while(recv(adaptor_threads->self_pipe[0],b,sizeof(b), 0)==sizeof(b)){}
-       }
+        else if (rc < 0) {
+            LOG_ERROR(LOGCALLBACK(zh), ("select() failed %d [%d].", rc, WSAGetLastError()));
+
+            // Clear interest events for zookeeper_process if select() fails.
+            interest = 0;
+        }
+
 #endif
         // dispatch zookeeper events
         rc = zookeeper_process(zh, interest);

src/c/src/zk_adaptor.h

@@ -230,7 +230,8 @@ struct _zhandle {
     clientid_t client_id;               // client-id
     long long last_zxid;                // last zookeeper ID
     auth_list_head_t auth_h;            // authentication data list
-    log_callback_fn log_callback;       // Callback for logging (falls back to logging to stderr)  
+    log_callback_fn log_callback;       // Callback for logging (falls back to logging to stderr)
+    int io_count;			// counts the number of iterations of do_io
 
     // Primer storage
     struct _buffer_list primer_buffer;  // The buffer used for the handshake at the start of a connection

src/c/src/zookeeper.c

@@ -1077,6 +1077,7 @@ static zhandle_t *zookeeper_init_internal(const char *host, watcher_fn watcher,
     } else {
         memset(&zh->client_id, 0, sizeof(zh->client_id));
     }
+    zh->io_count = 0;
     zh->primer_buffer.buffer = zh->primer_storage_buffer;
     zh->primer_buffer.curr_offset = 0;
     zh->primer_buffer.len = sizeof(zh->primer_storage_buffer);

src/c/tests/TestClient.cc

@@ -338,6 +338,20 @@ public:
         zrc = zoo_delete(zh, "/mytest/test1", -1);
         zookeeper_close(zh);
     }
+
+    void testBadDescriptor() {
+        int zrc = 0;
+        watchctx_t *ctx;
+        zhandle_t *zh = zookeeper_init(hostPorts, NULL, 10000, 0, ctx, 0);
+        sleep(1);
+        zh->io_count = 0;
+        //close socket
+        close(zh->fd);
+        sleep(1);
+        //Check that doIo isn't spinning
+        CPPUNIT_ASSERT(zh->io_count < 2);
+        zookeeper_close(zh);
+    }
     
 
     void testPing()