ManualExecutor.cpp 2.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  1. /*
  2. * Copyright 2014-present Facebook, Inc.
  3. *
  4. * Licensed under the Apache License, Version 2.0 (the "License");
  5. * you may not use this file except in compliance with the License.
  6. * You may obtain a copy of the License at
  7. *
  8. * http://www.apache.org/licenses/LICENSE-2.0
  9. *
  10. * Unless required by applicable law or agreed to in writing, software
  11. * distributed under the License is distributed on an "AS IS" BASIS,
  12. * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13. * See the License for the specific language governing permissions and
  14. * limitations under the License.
  15. */
  16. #include <folly/executors/ManualExecutor.h>
  17. #include <string.h>
  18. #include <string>
  19. #include <tuple>
  20. namespace folly {
  21. ManualExecutor::~ManualExecutor() {
  22. drain();
  23. }
  24. void ManualExecutor::add(Func callback) {
  25. std::lock_guard<std::mutex> lock(lock_);
  26. funcs_.emplace(std::move(callback));
  27. sem_.post();
  28. }
  29. size_t ManualExecutor::run() {
  30. size_t count;
  31. size_t n;
  32. Func func;
  33. {
  34. std::lock_guard<std::mutex> lock(lock_);
  35. while (!scheduledFuncs_.empty()) {
  36. auto& sf = scheduledFuncs_.top();
  37. if (sf.time > now_) {
  38. break;
  39. }
  40. funcs_.emplace(sf.moveOutFunc());
  41. scheduledFuncs_.pop();
  42. }
  43. n = funcs_.size();
  44. }
  45. for (count = 0; count < n; count++) {
  46. {
  47. std::lock_guard<std::mutex> lock(lock_);
  48. if (funcs_.empty()) {
  49. break;
  50. }
  51. // Balance the semaphore so it doesn't grow without bound
  52. // if nobody is calling wait().
  53. // This may fail (with EAGAIN), that's fine.
  54. sem_.tryWait();
  55. func = std::move(funcs_.front());
  56. funcs_.pop();
  57. }
  58. func();
  59. }
  60. return count;
  61. }
  62. size_t ManualExecutor::drain() {
  63. size_t tasksRun = 0;
  64. size_t tasksForSingleRun = 0;
  65. while ((tasksForSingleRun = run()) != 0) {
  66. tasksRun += tasksForSingleRun;
  67. }
  68. return tasksRun;
  69. }
  70. void ManualExecutor::wait() {
  71. while (true) {
  72. {
  73. std::lock_guard<std::mutex> lock(lock_);
  74. if (!funcs_.empty()) {
  75. break;
  76. }
  77. }
  78. sem_.wait();
  79. }
  80. }
  81. void ManualExecutor::advanceTo(TimePoint const& t) {
  82. if (t > now_) {
  83. now_ = t;
  84. }
  85. run();
  86. }
  87. } // namespace folly