1
// Copyright (c) 2017-2022 Cloudflare, Inc.
2
// Licensed under the Apache 2.0 license found in the LICENSE file or at:
3
//     https://opensource.org/licenses/Apache-2.0
4
 
5
#include "wait-list.h"
6
 
7
#include <kj/test.h>
8
#include <kj/thread.h>
9
 
10
namespace workerd {
11
namespace {
12
 
13
KJ_TEST("CrossThreadWaitList") {
14
  auto doTest = [](const CrossThreadWaitList& list) {
15
    kj::MutexGuarded<uint> ready;
16
 
17
    auto threadFunc = [&]() noexcept {
18
      kj::EventLoop loop;
19
      kj::WaitScope ws(loop);
20
 
21
      auto promise1 = list.addWaiter();
22
      auto promise2 = list.addWaiter();
23
 
24
      KJ_ASSERT(!promise1.poll(ws));
25
      KJ_ASSERT(!promise2.poll(ws));
26
      KJ_ASSERT(!list.isDone());
27
 
28
      (*ready.lockExclusive())++;
29
 
30
      promise1.wait(ws);
31
      promise2.wait(ws);
32
 
33
      KJ_ASSERT(list.isDone());
34
    };
35
 
36
    kj::Thread waiter1(threadFunc);
37
    kj::Thread waiter2(threadFunc);
38
    kj::Thread waiter3(threadFunc);
39
 
40
    kj::Thread sender([&]() {
41
      ready.when([](uint val) { return val == 3; }, [&](uint) {});
42
      list.fulfill();
43
    });
44
  };
45
 
46
  {
47
    CrossThreadWaitList list;
48
    doTest(list);
49
  }
50
  {
51
    CrossThreadWaitList list({.useThreadLocalOptimization = true});
52
    doTest(list);
53
  }
54
}
55
 
56
KJ_TEST("CrossThreadWaitList exceptions") {
57
  auto doTest = [](const CrossThreadWaitList& list) {
58
    kj::MutexGuarded<uint> ready;
59
 
60
    auto threadFunc = [&]() noexcept {
61
      kj::EventLoop loop;
62
      kj::WaitScope ws(loop);
63
 
64
      auto promise1 = list.addWaiter();
65
      auto promise2 = list.addWaiter();
66
 
67
      KJ_ASSERT(!promise1.poll(ws));
68
      KJ_ASSERT(!promise2.poll(ws));
69
      KJ_ASSERT(!list.isDone());
70
 
71
      (*ready.lockExclusive())++;
72
 
73
      promise1
74
          .then([]() { KJ_FAIL_REQUIRE("didn't throw"); }, [](kj::Exception&& e) {
75
        KJ_ASSERT(e.getDescription() == "foo");
76
      }).wait(ws);
77
      promise2
78
          .then([]() { KJ_FAIL_REQUIRE("didn't throw"); }, [](kj::Exception&& e) {
79
        KJ_ASSERT(e.getDescription() == "foo");
80
      }).wait(ws);
81
 
82
      KJ_ASSERT(list.isDone());
83
    };
84
 
85
    kj::Thread waiter1(threadFunc);
86
    kj::Thread waiter2(threadFunc);
87
    kj::Thread waiter3(threadFunc);
88
 
89
    kj::Thread sender([&]() {
90
      ready.when([](uint val) { return val == 3; }, [&](uint) {});
91
      list.reject(KJ_EXCEPTION(FAILED, "foo"));
92
    });
93
  };
94
 
95
  {
96
    CrossThreadWaitList list;
97
    doTest(list);
98
  }
99
  {
100
    CrossThreadWaitList list({.useThreadLocalOptimization = true});
101
    doTest(list);
102
  }
103
}
104
 
105
}  // namespace
106
}  // namespace workerd