fix worker leak when oom

deps/oblib/src/lib/thread/thread.cpp

@@ -61,7 +61,8 @@ Thread::Thread(Threads *threads, int64_t idx, int64_t stack_size)
       tid_before_stop_(0),
       tid_(0),
       thread_list_node_(this),
-      cpu_time_(0)
+      cpu_time_(0),
+      create_ret_(OB_NOT_RUNNING)
 {}
 
 Thread::~Thread()
@@ -98,9 +99,14 @@ int Thread::start()
     if (pret == 0) {
       stop_ = false;
       pret = pthread_create(&pth_, &attr, __th_start, this);
+      while (ATOMIC_LOAD(&create_ret_) == OB_NOT_RUNNING) {
+        sched_yield();
+      }
       if (pret != 0) {
         LOG_ERROR("pthread create failed", K(pret), K(errno));
         pth_ = 0;
+      } else if (OB_FAIL(create_ret_)) {
+        LOG_ERROR("thread create failed", K(create_ret_));
       }
     }
     if (0 != pret) {
@@ -320,6 +326,7 @@ void* Thread::__th_start(void *arg)
         WITH_CONTEXT(*mem_context) {
           try {
             in_try_stmt = true;
+            ATOMIC_STORE(&th->create_ret_, OB_SUCCESS);
             th->run();
             in_try_stmt = false;
           } catch (OB_BASE_EXCEPTION &except) {
@@ -327,6 +334,10 @@ void* Thread::__th_start(void *arg)
             _LOG_ERROR("Exception caught!!! errno = %d, exception info = %s", except.get_errno(), except.what());
             ret = OB_ERR_UNEXPECTED;
             in_try_stmt = false;
+            if (1 == th->threads_->get_thread_count() && !th->has_set_stop()) {
+              LOG_WARN("thread exit by itself without set_stop", K(ret));
+              th->threads_->stop();
+            }
           }
         }
       }
@@ -335,7 +346,6 @@ void* Thread::__th_start(void *arg)
       }
     }
   }
-
   ATOMIC_FAA(&total_thread_count_, -1);
   return nullptr;
 }

deps/oblib/src/lib/thread/thread.h

@@ -186,6 +186,7 @@ private:
   int64_t tid_;
   ThreadListNode thread_list_node_;
   int64_t cpu_time_;
+  int create_ret_;
 };
 
 OB_INLINE bool Thread::has_set_stop() const

deps/oblib/src/lib/thread/threads.h

@@ -103,8 +103,8 @@ public:
     }
     return pth;
   }
-protected:
   int64_t get_thread_count() const { return n_threads_; }
+protected:
   uint64_t get_thread_idx() const { return thread_idx_; }
   void set_thread_idx(int64_t idx) { thread_idx_ = idx; }