[util] Import ThreadPool and Thread from KUDU

be/src/common/status.h

@@ -67,7 +67,7 @@ class Status {
         return Status(TStatusCode::INVALID_ARGUMENT, msg, precise_code, msg2);
     }
     static Status MinimumReservationUnavailable(const Slice& msg, int16_t precise_code = 1, const Slice& msg2 = Slice()) {
-        return Status(TStatusCode::INVALID_ARGUMENT, msg, precise_code, msg2);
+        return Status(TStatusCode::MINIMUM_RESERVATION_UNAVAILABLE, msg, precise_code, msg2);
     }
     static Status Corruption(const Slice& msg, int16_t precise_code = 1, const Slice& msg2 = Slice()) {
         return Status(TStatusCode::CORRUPTION, msg, precise_code, msg2);
@@ -126,6 +126,21 @@ class Status {
     static Status TooManyTasks(const Slice& msg, int16_t precise_code = 1, const Slice& msg2 = Slice()) {
         return Status(TStatusCode::TOO_MANY_TASKS, msg, precise_code, msg2);
     }
+    static Status ServiceUnavailable(const Slice& msg,
+                                     int16_t precise_code = -1,
+                                     const Slice& msg2 = Slice()) {
+        return Status(TStatusCode::SERVICE_UNAVAILABLE, msg, precise_code, msg2);
+    }
+    static Status Uninitialized(const Slice& msg,
+                                int16_t precise_code = -1,
+                                const Slice& msg2 = Slice()) {
+        return Status(TStatusCode::UNINITIALIZED, msg, precise_code, msg2);
+    }
+    static Status Aborted(const Slice& msg,
+                          int16_t precise_code = -1,
+                          const Slice& msg2 = Slice()) {
+        return Status(TStatusCode::ABORTED, msg, precise_code, msg2);
+    }
 
     bool ok() const { return _state == nullptr; }
 
@@ -137,6 +152,18 @@ class Status {
     bool is_already_exist() const { return code() == TStatusCode::ALREADY_EXIST; }
     bool is_io_error() const {return code() == TStatusCode::IO_ERROR; }
 
+    /// @return @c true iff the status indicates Uninitialized.
+    bool is_uninitialized() const { return code() == TStatusCode::UNINITIALIZED; }
+
+    // @return @c true iff the status indicates an Aborted error.
+    bool is_aborted() const { return code() == TStatusCode::ABORTED; }
+
+    /// @return @c true iff the status indicates an InvalidArgument error.
+    bool is_invalid_argument() const { return code() == TStatusCode::INVALID_ARGUMENT; }
+
+    // @return @c true iff the status indicates ServiceUnavailable.
+    bool is_service_unavailable() const { return code() == TStatusCode::SERVICE_UNAVAILABLE; }
+
     // Convert into TStatus. Call this if 'status_container' contains an optional
     // TStatus field named 'status'. This also sets __isset.status.
     template <typename T>

be/src/util/CMakeLists.txt

@@ -92,6 +92,8 @@ set(UTIL_FILES
   monotime.cpp
   mutex.cpp
   condition_variable.cpp
+  thread.cpp
+  threadpool.cpp
 )
 
 if (WITH_MYSQL)

be/src/util/barrier.h

@@ -0,0 +1,70 @@
+// 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.
+
+#ifndef DORIS_BE_SRC_UTIL_BARRIER_H
+#define DORIS_BE_SRC_UTIL_BARRIER_H
+
+#include "gutil/macros.h"
+#include "olap/olap_define.h"
+#include "util/condition_variable.h"
+#include "util/mutex.h"
+
+namespace doris {
+
+// Implementation of pthread-style Barriers.
+class Barrier {
+public:
+    // Initialize the barrier with the given initial count.
+    explicit Barrier(int count) :
+        _cond(&_mutex),
+        _count(count),
+        _initial_count(count) {
+            DCHECK_GT(count, 0);
+        }
+
+    ~Barrier() {}
+
+    // wait until all threads have reached the barrier.
+    // Once all threads have reached the barrier, the barrier is reset
+    // to the initial count.
+    void wait() {
+        MutexLock l(&_mutex);
+        if (--_count == 0) {
+            _count = _initial_count;
+            _cycle_count++;
+            _cond.notify_all();
+            return;
+        }
+
+        int initial_cycle = _cycle_count;
+        while (_cycle_count == initial_cycle) {
+            _cond.wait();
+        }
+    }
+
+private:
+    Mutex _mutex;
+    ConditionVariable _cond;
+    int _count;
+    uint32_t _cycle_count = 0;
+    const int _initial_count;
+    DISALLOW_COPY_AND_ASSIGN(Barrier);
+};
+
+#endif //DORIS_BE_SRC_UTIL_BARRIER_H
+
+} // namespace doris

be/src/util/question

@@ -0,0 +1,3 @@
+Regression race of lock
+timeout of condition_variable
+deadlock