From f7931a7a091b95e237eb0f06857c68e5e66c1d1b Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Fri, 21 Aug 2026 01:35:06 -0700 Subject: [PATCH] Return when C client worker threads fail to start Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../zookeeper-client-c/src/mt_adaptor.c | 40 +++++++-- .../tests/TestZookeeperInit.cc | 82 +++++++++++++++++++ 2 files changed, 116 insertions(+), 6 deletions(-) diff --git a/zookeeper-client/zookeeper-client-c/src/mt_adaptor.c b/zookeeper-client/zookeeper-client-c/src/mt_adaptor.c index 174701c7358..b0d2cf095da 100644 --- a/zookeeper-client/zookeeper-client-c/src/mt_adaptor.c +++ b/zookeeper-client/zookeeper-client-c/src/mt_adaptor.c @@ -199,7 +199,8 @@ 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); @@ -207,9 +208,10 @@ void notify_thread_ready(zhandle_t* zh) } -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); diff --git a/zookeeper-client/zookeeper-client-c/tests/TestZookeeperInit.cc b/zookeeper-client/zookeeper-client-c/tests/TestZookeeperInit.cc index d5fc12c5ded..f9623502986 100644 --- a/zookeeper-client/zookeeper-client-c/tests/TestZookeeperInit.cc +++ b/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}};