YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

//

// Copyright (c) YugaByte, Inc.

//

// Licensed 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 YB_RPC_THREAD_POOL_H

#define YB_RPC_THREAD_POOL_H

#include <memory>

#include <string>

#include "yb/gutil/port.h"

#include "yb/util/tostring.h"

namespace yb {

class Status;

class Thread;

namespace rpc {

class ThreadPoolTask {

 public:

  // Invoked in thread pool

  virtual void Run() = 0;

  // When thread pool done with task, i.e. it completed or failed, it invokes Done

  virtual void Done(const Status& status) = 0;

 protected:

  ~ThreadPoolTask() {}

};

template <class F, class Base = ThreadPoolTask>

class FunctorThreadPoolTask : public Base {

 public:

  explicit FunctorThreadPoolTask(const F& f) : f_(f) {}

  explicit FunctorThreadPoolTask(F&& f) : f_(std::move(f)) {}

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand26ThreadPoolTest_Strand_Test8TestBodyEvE3$_3NS0_10StrandTaskEEC2EOS4_

Line

Count

Source

49

100

  explicit FunctorThreadPoolTask(F&& f) : f_(std::move(f)) {}

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand34ThreadPoolTest_StrandShutdown_Test8TestBodyEvE3$_4NS0_10StrandTaskEEC2EOS4_

Line

Count

Source

49

1

  explicit FunctorThreadPoolTask(F&& f) : f_(std::move(f)) {}

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand48ThreadPoolTest_StrandShutdownAndDestroyRace_Test8TestBodyEvE3$_7NS0_10StrandTaskEEC2EOS4_

Line

Count

Source

49

20

  explicit FunctorThreadPoolTask(F&& f) : f_(std::move(f)) {}

  virtual ~FunctorThreadPoolTask() = default;

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand26ThreadPoolTest_Strand_Test8TestBodyEvE3$_3NS0_10StrandTaskEED2Ev

Line

Count

Source

51

100

  virtual ~FunctorThreadPoolTask() = default;

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand34ThreadPoolTest_StrandShutdown_Test8TestBodyEvE3$_4NS0_10StrandTaskEED2Ev

Line

Count

Source

51

1

  virtual ~FunctorThreadPoolTask() = default;

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand48ThreadPoolTest_StrandShutdownAndDestroyRace_Test8TestBodyEvE3$_7NS0_10StrandTaskEED2Ev

Line

Count

Source

51

20

  virtual ~FunctorThreadPoolTask() = default;

 private:

  void Run() override {

    f_();

  }

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand26ThreadPoolTest_Strand_Test8TestBodyEvE3$_3NS0_10StrandTaskEE3RunEv

Line

Count

Source

54

100

  void Run() override {

55

100

    f_();

56

100

  }

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand34ThreadPoolTest_StrandShutdown_Test8TestBodyEvE3$_4NS0_10StrandTaskEE3RunEv

Line

Count

Source

54

1

  void Run() override {

55

1

    f_();

56

1

  }

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand48ThreadPoolTest_StrandShutdownAndDestroyRace_Test8TestBodyEvE3$_7NS0_10StrandTaskEE3RunEv

Line

Count

Source

54

10

  void Run() override {

55

10

    f_();

56

10

  }

  void Done(const Status& status) override {

    delete this;

  }

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand26ThreadPoolTest_Strand_Test8TestBodyEvE3$_3NS0_10StrandTaskEE4DoneERKNS_6StatusE

Line

Count

Source

58

100

  void Done(const Status& status) override {

59

100

    delete this;

60

100

  }

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand34ThreadPoolTest_StrandShutdown_Test8TestBodyEvE3$_4NS0_10StrandTaskEE4DoneERKNS_6StatusE

Line

Count

Source

58

1

  void Done(const Status& status) override {

59

1

    delete this;

60

1

  }

thread_pool-test.cc:_ZN2yb3rpc21FunctorThreadPoolTaskIZNS0_6strand48ThreadPoolTest_StrandShutdownAndDestroyRace_Test8TestBodyEvE3$_7NS0_10StrandTaskEE4DoneERKNS_6StatusE

Line

Count

Source

58

20

  void Done(const Status& status) override {

59

20

    delete this;

60

20

  }

  F f_;

};

template <class F>

FunctorThreadPoolTask<F>* MakeFunctorThreadPoolTask(const F& f) {

  return new FunctorThreadPoolTask<F>(f);

}

template <class F>

FunctorThreadPoolTask<F>* MakeFunctorThreadPoolTask(F&& f) {

  return new FunctorThreadPoolTask<F>(std::move(f));

}

struct ThreadPoolOptions {

  std::string name;

  size_t queue_limit;

  size_t max_workers;

  std::string ToString() const {

    return YB_STRUCT_TO_STRING(name, queue_limit, max_workers);

  }

};

class ThreadPool {

 public:

  explicit ThreadPool(ThreadPoolOptions options);

  template <class... Args>

  explicit ThreadPool(Args&&... args)

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

  }

_ZN2yb3rpc10ThreadPoolC1IJRA11_KcRKmmEEEDpOT_

Line

Count

Source

91

1

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

92

93

1

  }

_ZN2yb3rpc10ThreadPoolC1IJRA5_KcRKmS7_EEEDpOT_

Line

Count

Source

91

10

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

92

93

10

  }

_ZN2yb3rpc10ThreadPoolC1IJRA9_KcRKmS7_EEEDpOT_

Line

Count

Source

91

82

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

92

93

82

  }

_ZN2yb3rpc10ThreadPoolC1IJRA19_KcRyS6_EEEDpOT_

Line

Count

Source

91

4.72k

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

92

93

4.72k

  }

_ZN2yb3rpc10ThreadPoolC1IJNSt3__112basic_stringIcNS3_11char_traitsIcEENS3_9allocatorIcEEEERKmSB_EEEDpOT_

Line

Count

Source

91

11.2k

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

92

93

11.2k

  }

_ZN2yb3rpc10ThreadPoolC1IJRKNSt3__112basic_stringIcNS3_11char_traitsIcEENS3_9allocatorIcEEEERKmSD_EEEDpOT_

Line

Count

Source

91

21.3k

      : ThreadPool(ThreadPoolOptions{std::forward<Args>(args)...}) {

92

93

21.3k

  }

  ~ThreadPool();

  ThreadPool(ThreadPool&& rhs) noexcept;

  ThreadPool& operator=(ThreadPool&& rhs) noexcept;

  const ThreadPoolOptions& options() const;

  bool Enqueue(ThreadPoolTask* task);

  template <class F>

  void EnqueueFunctor(const F& f) {

    Enqueue(MakeFunctorThreadPoolTask(f));

  }

  template <class F>

  void EnqueueFunctor(F&& f) {

    Enqueue(MakeFunctorThreadPoolTask(std::move(f)));

  }

  void Shutdown();

  static bool IsCurrentThreadRpcWorker();

  bool Owns(Thread* thread);

  bool OwnsThisThread();

 private:

  class Impl;

  std::unique_ptr<Impl> impl_;

};

} // namespace rpc

} // namespace yb

#endif // YB_RPC_THREAD_POOL_H