1//
2// Copyright (c) 2022 Klemens Morgenstern (klemens.morgenstern@gmx.net)
3//
4// Distributed under the Boost Software License, Version 1.0. (See accompanying
5// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6//
7
8#ifndef BOOST_COBALT_CHANNEL_HPP
9#define BOOST_COBALT_CHANNEL_HPP
10
11#include <boost/cobalt/this_thread.hpp>
12#include <boost/cobalt/unique_handle.hpp>
13#include <boost/cobalt/detail/util.hpp>
14
15#include <boost/asio/cancellation_signal.hpp>
16#include <boost/asio/cancellation_type.hpp>
17#include <boost/circular_buffer.hpp>
18#include <boost/config.hpp>
19#include <boost/intrusive/list.hpp>
20#include <boost/variant2/variant.hpp>
21
22#include <optional>
23
24namespace boost::cobalt
25{
26
27template<typename T>
28struct channel_reader;
29
30// tag::outline[]
31template<typename T>
32struct channel
33{
34 // end::outline[]
35#if defined(BOOST_COBALT_NO_PMR)
36 channel(std::size_t limit = 0u,
37 executor executor = this_thread::get_executor());
38#else
39 // tag::outline[]
40 // create a channel with a buffer limit, executor & resource.
41 explicit
42 channel(std::size_t limit = 0u,
43 executor executor = this_thread::get_executor(),
44 pmr::memory_resource * resource = this_thread::get_default_resource());
45 // end::outline[]
46#endif
47 // tag::outline[]
48 // not movable.
49 channel(channel && rhs) noexcept = delete;
50 channel & operator=(channel && lhs) noexcept = delete;
51
52 using executor_type = executor;
53 const executor_type & get_executor();
54
55 // Closes the channel
56 ~channel();
57 bool is_open() const;
58 // close the operation, will cancel all pending ops, too
59 void close();
60
61 // end::outline[]
62 private:
63#if !defined(BOOST_COBALT_NO_PMR)
64 boost::circular_buffer<T, pmr::polymorphic_allocator<T>> buffer_;
65#else
66 boost::circular_buffer<T> buffer_;
67#endif
68 executor_type executor_;
69 bool is_closed_{false};
70
71 struct read_op : intrusive::list_base_hook<intrusive::link_mode<intrusive::auto_unlink> >
72 {
73 channel * chn;
74 boost::source_location loc;
75 bool cancelled = false;
76 std::optional<T> direct{};
77 asio::cancellation_slot cancel_slot{};
78 unique_handle<void> awaited_from{nullptr};
79 void (*begin_transaction)(void*) = nullptr;
80
81 void transactional_unlink()
82 {
83 if (begin_transaction)
84 begin_transaction(awaited_from.get());
85 this->unlink();
86 }
87
88 struct cancel_impl;
89 bool await_ready() { return !chn->buffer_.empty(); }
90 template<typename Promise>
91 BOOST_NOINLINE
92 std::coroutine_handle<void> await_suspend(std::coroutine_handle<Promise> h);
93 T await_resume();
94 std::tuple<system::error_code, T> await_resume(const struct as_tuple_tag & );
95 system::result<T> await_resume(const struct as_result_tag &);
96 explicit operator bool() const {return chn && chn->is_open();}
97 };
98
99 struct write_op : intrusive::list_base_hook<intrusive::link_mode<intrusive::auto_unlink> >
100 {
101 channel * chn;
102 variant2::variant<T*, const T*> ref;
103 boost::source_location loc;
104 bool cancelled = false, direct = false;
105 asio::cancellation_slot cancel_slot{};
106
107 unique_handle<void> awaited_from{nullptr};
108 void (*begin_transaction)(void*) = nullptr;
109
110 void transactional_unlink()
111 {
112 if (begin_transaction)
113 begin_transaction(awaited_from.get());
114 this->unlink();
115 }
116
117 struct cancel_impl;
118
119 bool await_ready() { return !chn->buffer_.full(); }
120 template<typename Promise>
121 BOOST_NOINLINE
122 std::coroutine_handle<void> await_suspend(std::coroutine_handle<Promise> h);
123 void await_resume();
124 std::tuple<system::error_code> await_resume(const struct as_tuple_tag & );
125 system::result<void> await_resume(const struct as_result_tag &);
126 explicit operator bool() const {return chn && chn->is_open();}
127 };
128
129 boost::intrusive::list<read_op, intrusive::constant_time_size<false> > read_queue_;
130 boost::intrusive::list<write_op, intrusive::constant_time_size<false> > write_queue_;
131 public:
132 read_op read(const boost::source_location & loc = BOOST_CURRENT_LOCATION) {return read_op{{}, this, loc}; }
133 write_op write(const T && value, const boost::source_location & loc = BOOST_CURRENT_LOCATION)
134 {
135 return write_op{{}, this, &value, loc};
136 }
137 write_op write(const T & value, const boost::source_location & loc = BOOST_CURRENT_LOCATION)
138 {
139 return write_op{{}, this, &value, loc};
140 }
141 write_op write( T && value, const boost::source_location & loc = BOOST_CURRENT_LOCATION)
142 {
143 return write_op{{}, this, &value, loc};
144 }
145 write_op write( T & value, const boost::source_location & loc = BOOST_CURRENT_LOCATION)
146 {
147 return write_op{{}, this, &value, loc};
148 }
149 /*
150 // tag::outline[]
151 // an awaitable that yields T
152 using __read_op__ = __unspecified__;
153
154 // an awaitable that yields void
155 using __write_op__ = __unspecified__;
156
157 // read a value to a channel
158 __read_op__ read();
159
160 // write a value to the channel
161 __write_op__ write(const T && value);
162 __write_op__ write(const T & value);
163 __write_op__ write( T && value);
164 __write_op__ write( T & value);
165
166 // write a value to the channel if T is void
167 __write_op__ write(); // end::outline[]
168 */
169 // tag::outline[]
170
171};
172// end::outline[]
173
174template<>
175struct channel<void>
176{
177 explicit
178 channel(std::size_t limit = 0u,
179 executor executor = this_thread::get_executor())
180 : limit_(limit), executor_(executor) {}
181 channel(channel &&) noexcept = delete;
182 channel & operator=(channel && lhs) noexcept = delete;
183
184 using executor_type = executor;
185 const executor_type & get_executor() {return executor_;}
186
187 BOOST_COBALT_DECL ~channel();
188
189 bool is_open() const {return !is_closed_;}
190 BOOST_COBALT_DECL void close();
191
192 private:
193 std::size_t limit_;
194 std::size_t n_{0u};
195 executor_type executor_;
196 bool is_closed_{false};
197
198 struct read_op : intrusive::list_base_hook<intrusive::link_mode<intrusive::auto_unlink> >
199 {
200 channel * chn;
201 boost::source_location loc;
202 bool cancelled = false, direct = false;
203 asio::cancellation_slot cancel_slot{};
204 unique_handle<void> awaited_from{nullptr};
205 void (*begin_transaction)(void*) = nullptr;
206
207 void transactional_unlink()
208 {
209 if (begin_transaction)
210 begin_transaction(awaited_from.get());
211 this->unlink();
212 }
213
214 struct cancel_impl;
215 bool await_ready() { return (chn->n_ > 0); }
216 template<typename Promise>
217 BOOST_NOINLINE
218 std::coroutine_handle<void> await_suspend(std::coroutine_handle<Promise> h);
219 BOOST_COBALT_DECL void await_resume();
220 BOOST_COBALT_DECL std::tuple<system::error_code> await_resume(const struct as_tuple_tag & );
221 BOOST_COBALT_DECL system::result<void> await_resume(const struct as_result_tag &);
222 explicit operator bool() const {return chn && chn->is_open();}
223 };
224
225 struct write_op : intrusive::list_base_hook<intrusive::link_mode<intrusive::auto_unlink> >
226 {
227 channel * chn;
228 boost::source_location loc;
229 bool cancelled = false, direct = false;
230 asio::cancellation_slot cancel_slot{};
231 unique_handle<void> awaited_from{nullptr};
232 void (*begin_transaction)(void*) = nullptr;
233
234 void transactional_unlink()
235 {
236 if (begin_transaction)
237 begin_transaction(awaited_from.get());
238 this->unlink();
239 }
240
241 struct cancel_impl;
242 bool await_ready()
243 {
244 return chn->n_ < chn->limit_;
245 }
246
247 template<typename Promise>
248 BOOST_NOINLINE
249 std::coroutine_handle<void> await_suspend(std::coroutine_handle<Promise> h);
250
251 BOOST_COBALT_DECL void await_resume();
252 BOOST_COBALT_DECL std::tuple<system::error_code> await_resume(const struct as_tuple_tag & );
253 BOOST_COBALT_DECL system::result<void> await_resume(const struct as_result_tag &);
254 explicit operator bool() const {return chn && chn->is_open();}
255 };
256
257 boost::intrusive::list<read_op, intrusive::constant_time_size<false> > read_queue_;
258 boost::intrusive::list<write_op, intrusive::constant_time_size<false> > write_queue_;
259 public:
260 read_op read(const boost::source_location & loc = BOOST_CURRENT_LOCATION) {return read_op{{}, .chn: this, .loc: loc}; }
261 write_op write(const boost::source_location & loc = BOOST_CURRENT_LOCATION) {return write_op{{}, .chn: this, .loc: loc}; }
262};
263
264template<typename T>
265struct channel_reader
266{
267 channel_reader(channel<T> & chan,
268 const boost::source_location & loc = BOOST_CURRENT_LOCATION) : chan_(&chan), loc_(loc) {}
269
270 auto operator co_await ()
271 {
272 return chan_->read(loc_);
273 }
274
275 explicit operator bool () const {return chan_ && chan_->is_open();}
276
277 private:
278 channel<T> * chan_;
279 boost::source_location loc_;
280};
281
282}
283
284#include <boost/cobalt/impl/channel.hpp>
285
286#endif //BOOST_COBALT_CHANNEL_HPP
287

source code of boost/libs/cobalt/include/boost/cobalt/channel.hpp