1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
|
Futures and promises
--------------------
A *future* is a result of a computation that may not be available yet.
Examples include:
* a data buffer that we are reading from the network
* the expiration of a timer
* the completion of a disk write
* the result computation that requires the values from
one or more other futures.
a *promise* is an object or function that provides you with a future,
with the expectation that it will fulfill the future.
Promises and futures simplify asynchronous programming since they decouple
the event producer (the promise) and the event consumer (whoever uses the
future). Whether the promise is fulfilled before the future is consumed,
or vice versa, does not change the outcome of the code.
Consuming a future
------------------
You consume a future by using its *then()* method, providing it with a
callback (typically a lambda). For example, consider the following
operation:
```C++
future<int> get(); // promises an int will be produced eventually
future<> put(int) // promises to store an int
void f() {
get().then([] (int value) {
put(value + 1).then([] {
std::cout << "value stored successfully\n";
});
});
}
```
Here, we initiate a *get()* operation, requesting that when it completes, a
*put()* operation will be scheduled with an incremented value. We also
request that when the *put()* completes, some text will be printed out.
Chaining futures
----------------
If a *then()* lambda returns a future (call it x), then that *then()*
will return a future (call it y) that will receive the same value. This
removes the need for nesting lambda blocks; for example the code above
could be rewritten as:
```C++
future<int> get(); // promises an int will be produced eventually
future<> put(int) // promises to store an int
void f() {
get().then([] (int value) {
return put(value + 1);
}).then([] {
std::cout << "value stored successfully\n";
});
}
```
Loops
-----
Loops are achieved with a tail call; for example:
```C++
future<int> get(); // promises an int will be produced eventually
future<> put(int) // promises to store an int
future<> loop_to(int end) {
if (value == end) {
return make_ready_future<>();
}
get().then([end] (int value) {
return put(value + 1);
}).then([end] {
return loop_to(end);
});
}
```
The *make_ready_future()* function returns a future that is already
available --- corresponding to the loop termination condition, where
no further I/O needs to take place.
Under the hood
--------------
When the loop above runs, both *then* method calls execute immediately
--- but without executing the bodies. What happens is the following:
1. `get()` is called, initiates the I/O operation, and allocates a
temporary structure (call it `f1`).
2. The first `then()` call chains its body to `f1` and allocates
another temporary structure, `f2`.
3. The second `then()` call chains its body to `f2`.
Again, all this runs immediately without waiting for anything.
After the I/O operation initiated by `get()` completes, it calls the
continuation stored in `f1`, calls it, and frees `f1`. The continuation
calls `put()`, which initiates the I/O operation required to perform
the store, and allocates a temporary object `f12`, and chains some glue
code to it.
After the I/O operation initiated by `put()` completes, it calls the
continuation associated with `f12`, which simply tells it to call the
continuation associated with `f2`. This continuation simply calls
`loop_to()`. Both `f12` and `f2` are freed. `loop_to()` then calls
`get()`, which starts the process all over again, allocating new versions
of `f1` and `f2`.
Handling exceptions
-------------------
If a `.then()` clause throws an exception, the scheduler will catch it
and cancel any dependent `.then()` clauses. If you want to trap the
exception, add a `.then_wrapped()` clause at the end:
```C++
future<buffer> receive();
request parse(buffer buf);
future<response> process(request req);
future<> send(response resp);
void f() {
receive().then([] (buffer buf) {
return process(parse(std::move(buf));
}).then([] (response resp) {
return send(std::move(resp));
}).then([] {
f();
}).then_wrapped([] (auto&& f) {
try {
f.get();
} catch (std::exception& e) {
// your handler goes here
}
});
}
```
The previous future is passed as a parameter to the lambda, and its value can
be inspected with `f.get()`. When the `get()` variable is called as a
function, it will re-throw the exception that aborted processing, and you can
then apply any needed error handling. It is essentially a transformation of
```C++
buffer receive();
request parse(buffer buf);
response process(request req);
void send(response resp);
void f() {
try {
while (true) {
auto req = parse(receive());
auto resp = process(std::move(req));
send(std::move(resp));
}
} catch (std::exception& e) {
// your handler goes here
}
}
```
Note, however, that the `.then_wrapped()` clause will be scheduled both when
exception occurs or not. Therefore, the mere fact that `.then_wrapped()` is
executed does not mean that an exception was thrown. Only the execution of the
catch block can guarantee that.
This is shown below:
```C++
future<my_type> receive();
void f() {
receive().then_wrapped([] (future<my_type> f) {
try {
my_type x = f.get();
return do_something(x);
} catch (std::exception& e) {
// your handler goes here
}
});
}
```
### Setup notes
SeaStar is a high performance framework and tuned to get the best
performance by default. As such, we're tuned towards polling vs interrupt
driven. Our assumption is that applications written for SeaStar will be
busy handling 100,000 IOPS and beyond. Polling means that each of our
cores will consume 100% cpu even when no work is given to it.
|
