Return when C client worker threads fail to start

Signed-off-by: 1fanwang <1fannnw@gmail.com>

Author: 1fanwang

zookeeper-client/zookeeper-client-c/src/mt_adaptor.c

@@ -199,17 +199,19 @@ void notify_thread_ready(zhandle_t* zh)
 {
     struct adaptor_threads* adaptor=zh->adaptor_priv;
     pthread_mutex_lock(&adaptor->lock);
-    adaptor->threadsToWait--;
+    if(adaptor->threadsToWait>0)
+        adaptor->threadsToWait--;
     pthread_cond_broadcast(&adaptor->cond);
     while(adaptor->threadsToWait>0) 
         pthread_cond_wait(&adaptor->cond,&adaptor->lock);
     pthread_mutex_unlock(&adaptor->lock);
 }
 
 
-void start_threads(zhandle_t* zh)
+int start_threads(zhandle_t* zh)
 {
     int rc = 0;
+    int io_started = 0;
     struct adaptor_threads* adaptor=zh->adaptor_priv;
     pthread_cond_init(&adaptor->cond,0);
     pthread_mutex_init(&adaptor->lock,0);
@@ -220,15 +222,34 @@ void start_threads(zhandle_t* zh)
     api_prolog(zh);
     LOG_DEBUG(LOGCALLBACK(zh), "starting threads...");
     rc=pthread_create(&adaptor->io, 0, do_io, zh);
-    assert("pthread_create() failed for the IO thread"&&!rc);
+    if(rc)
+        goto fail;
+    io_started=1;
     rc=pthread_create(&adaptor->completion, 0, do_completion, zh);
-    assert("pthread_create() failed for the completion thread"&&!rc);
+    if(rc)
+        goto fail;
     wait_for_others(zh);
-    api_epilog(zh, 0);    
+    api_epilog(zh, 0);
+    return 0;
+
+fail:
+    LOG_ERROR(LOGCALLBACK(zh), "pthread_create() failed: %d", rc);
+    if(io_started) {
+        zh->close_requested=1;
+        pthread_mutex_lock(&adaptor->lock);
+        adaptor->threadsToWait=0;
+        pthread_cond_broadcast(&adaptor->cond);
+        pthread_mutex_unlock(&adaptor->lock);
+        pthread_join(adaptor->io, 0);
+        zh->close_requested=0;
+    }
+    api_epilog(zh, 0);
+    return rc;
 }
 
 int adaptor_init(zhandle_t *zh)
 {
+    int rc;
     pthread_mutexattr_t recursive_mx_attr;
     struct adaptor_threads *adaptor_threads = calloc(1, sizeof(*adaptor_threads));
     if (!adaptor_threads) {
@@ -267,7 +288,12 @@ int adaptor_init(zhandle_t *zh)
     pthread_cond_init(&zh->sent_requests.cond,0);
     pthread_mutex_init(&zh->completions_to_process.lock,0);
     pthread_cond_init(&zh->completions_to_process.cond,0);
-    start_threads(zh);
+    rc=start_threads(zh);
+    if(rc) {
+        adaptor_destroy(zh);
+        errno=rc;
+        return -1;
+    }
     return 0;
 }
 
@@ -313,6 +339,8 @@ void adaptor_destroy(zhandle_t *zh)
     pthread_mutex_destroy(&zh->completions_to_process.lock);
     pthread_cond_destroy(&zh->completions_to_process.cond);
     pthread_mutex_destroy(&adaptor->zh_lock);
+    pthread_mutex_destroy(&adaptor->reconfig_lock);
+    pthread_mutex_destroy(&adaptor->watchers_lock);
 
     pthread_mutex_destroy(&zh->auth_h.lock);
 

zookeeper-client/zookeeper-client-c/tests/TestZookeeperInit.cc

@@ -33,11 +33,63 @@ class MockPthreadsNull;
 
 using namespace std;
 
+#ifdef THREADED
+class CheckedPthreadCreateFailure : public CheckedPthread
+{
+public:
+    explicit CheckedPthreadCreateFailure(int failOnCall):
+        pthreadCreateCounter(0),pthreadJoinCounter(0),
+        pthreadCondDestroyCounter(0),pthreadMutexDestroyCounter(0),
+        threadCreated(false),failOnCall_(failOnCall){}
+
+    int pthread_create(pthread_t *t, const pthread_attr_t *a,
+            void *(*f)(void *), void *d) override {
+        if (pthreadCreateCounter++ == failOnCall_)
+            return EAGAIN;
+        int rc=CheckedPthread::pthread_create(t,a,f,d);
+        if(!rc) {
+            thread=*t;
+            threadCreated=true;
+        }
+        return rc;
+    }
+
+    int pthread_join(pthread_t t, void **r) override {
+        pthreadJoinCounter++;
+        return CheckedPthread::pthread_join(t,r);
+    }
+
+    int pthread_cond_destroy(pthread_cond_t *c) override {
+        pthreadCondDestroyCounter++;
+        return CheckedPthread::pthread_cond_destroy(c);
+    }
+
+    int pthread_mutex_destroy(pthread_mutex_t *m) override {
+        pthreadMutexDestroyCounter++;
+        return CheckedPthread::pthread_mutex_destroy(m);
+    }
+
+    int pthreadCreateCounter;
+    int pthreadJoinCounter;
+    int pthreadCondDestroyCounter;
+    int pthreadMutexDestroyCounter;
+    pthread_t thread;
+    bool threadCreated;
+
+private:
+    int failOnCall_;
+};
+#endif
+
 class Zookeeper_init : public CPPUNIT_NS::TestFixture
 {
     CPPUNIT_TEST_SUITE(Zookeeper_init);
     CPPUNIT_TEST(testVersion);
     CPPUNIT_TEST(testBasic);
+#ifdef THREADED
+    CPPUNIT_TEST(testFirstThreadCreateFailure);
+    CPPUNIT_TEST(testSecondThreadCreateFailure);
+#endif
     CPPUNIT_TEST(testAddressResolution);
     CPPUNIT_TEST(testMultipleAddressResolution);
     CPPUNIT_TEST(testNullAddressString);
@@ -139,6 +191,36 @@ class Zookeeper_init : public CPPUNIT_NS::TestFixture
         CPPUNIT_ASSERT(MockPthreadsNull::isInitialized(&zh->completions_to_process.cond));
 #endif
     }
+#ifdef THREADED
+    void assertThreadCreateFailure(int failOnCall)
+    {
+        delete pthreadMock;
+        pthreadMock=0;
+        CheckedPthreadCreateFailure pthreadFailure(failOnCall);
+
+        errno=0;
+        zh=zookeeper_init("127.0.0.1:2121",watcher,10000,0,0,0);
+
+        CPPUNIT_ASSERT(zh==0);
+        CPPUNIT_ASSERT_EQUAL(EAGAIN,errno);
+        CPPUNIT_ASSERT_EQUAL(failOnCall+1,pthreadFailure.pthreadCreateCounter);
+        CPPUNIT_ASSERT_EQUAL(failOnCall,pthreadFailure.pthreadJoinCounter);
+        CPPUNIT_ASSERT_EQUAL(3,pthreadFailure.pthreadCondDestroyCounter);
+        CPPUNIT_ASSERT_EQUAL(9,pthreadFailure.pthreadMutexDestroyCounter);
+        if(pthreadFailure.threadCreated)
+            CPPUNIT_ASSERT(CheckedPthread::isDestroyed(pthreadFailure.thread));
+    }
+
+    void testFirstThreadCreateFailure()
+    {
+        assertThreadCreateFailure(0);
+    }
+
+    void testSecondThreadCreateFailure()
+    {
+        assertThreadCreateFailure(1);
+    }
+#endif
     void testAddressResolution()
     {
         const char EXPECTED_IPS[][4]={{127,0,0,1}};