@@ -27,7 +27,9 @@ bool _hasattr(object o, const char* name)
2727 return PyObject_HasAttrString (o.ptr (), name);
2828}
2929
30- void _sock_connect_cb (object pymod_socket, object fut, object sock, object addr)
30+ }
31+
32+ void event_loop::_sock_connect_cb (object pymod_socket, object fut, object sock, object addr)
3133{
3234 try
3335 {
@@ -61,11 +63,10 @@ void _sock_connect_cb(object pymod_socket, object fut, object sock, object addr)
6163 }
6264}
6365
64- void _sock_accept (event_loop& loop, object fut, object sock)
66+ void event_loop:: _sock_accept (event_loop& loop, object fut, object sock)
6567{
6668 int fd = extract<int >(sock.attr (" fileno" )());
67- object conn;
68- object address;
69+ object conn, address;
6970 try
7071 {
7172 object ret = sock.attr (" accept" )();
@@ -80,9 +81,7 @@ void _sock_accept(event_loop& loop, object fut, object sock)
8081 || PyErr_ExceptionMatches (PyExc_InterruptedError))
8182 {
8283 PyErr_Clear ();
83- loop.add_reader (fd, make_function (bind (
84- _sock_accept, boost::ref (loop), fut, sock),
85- default_call_policies (), boost::mpl::vector<void , object>()));
84+ loop._async_wait_fd (fd, bind (_sock_accept, boost::ref (loop), fut, sock), loop._write_key (fd));
8685 }
8786 else if (PyErr_ExceptionMatches (PyExc_SystemExit)
8887 || PyErr_ExceptionMatches (PyExc_KeyboardInterrupt))
@@ -94,12 +93,11 @@ void _sock_accept(event_loop& loop, object fut, object sock)
9493 PyErr_Clear ();
9594 fut.attr (" set_exception" )(std::current_exception ());
9695 }
97- }
98- }
99-
96+ }
10097}
10198
102- void event_loop::_add_reader_or_writer (int fd, object f, int key)
99+ template <typename F>
100+ void event_loop::_async_wait_fd (int fd, F f, int key)
103101{
104102 // add descriptor
105103 if (_descriptor_map.find (key) == _descriptor_map.end ())
@@ -110,30 +108,14 @@ void event_loop::_add_reader_or_writer(int fd, object f, int key)
110108 }
111109
112110 _descriptor_map.find (key)->second ->async_wait (boost::asio::posix::descriptor::wait_type::wait_read,
113- boost::asio::bind_executor (_strand, [key, f, loop= this ] (const boost::system::error_code& ec)
111+ boost::asio::bind_executor (_strand, [this , key, f ] (const boost::system::error_code& ec)
114112 {
115- // move descriptor
116- auto iter = loop->_descriptor_map .find (key);
117- if (iter != loop->_descriptor_map .end ())
118- {
119- iter->second ->release ();
120- loop->_descriptor_map .erase (iter);
121- }
122- loop->call_soon (f);
113+ _descriptor_map.erase (key);
114+ f ();
123115 }));
124116 return ;
125117}
126118
127- void event_loop::_remove_reader_or_writer (int key)
128- {
129- auto iter = _descriptor_map.find (key);
130- if (iter != _descriptor_map.end ())
131- {
132- iter->second ->release ();
133- _descriptor_map.erase (iter);
134- }
135- }
136-
137119void event_loop::call_later (double delay, object f)
138120{
139121 auto p_timer = std::make_shared<boost::asio::steady_timer>(_strand.context (),
@@ -156,13 +138,13 @@ object event_loop::sock_recv(object sock, size_t nbytes)
156138 int fd = extract<int >(sock.attr (" fileno" )());
157139 int fd_dup = dup (fd);
158140 object py_fut = _pymod_concurrent_future.attr (" Future" )();
159- add_reader (fd_dup, make_function (
160- [py_fut, nbytes, fd=fd_dup] (object obj) {
141+ _async_wait_fd (fd_dup,
142+ [py_fut, nbytes, fd=fd_dup] {
161143 std::vector<char > buffer (nbytes);
162144 read (fd, buffer.data (), nbytes);
163145 py_fut.attr (" set_result" )(object (handle<>(PyBytes_FromStringAndSize (buffer.data (), nbytes))));
164146 },
165- default_call_policies (), boost::mpl::vector< void , object>() ));
147+ _read_key (fd ));
166148 return py_fut;
167149}
168150
@@ -172,13 +154,13 @@ object event_loop::sock_recv_into(object sock, object buffer)
172154 int fd_dup = dup (fd);
173155 ssize_t nbytes = len (buffer);
174156 object py_fut = _pymod_concurrent_future.attr (" Future" )();
175- add_reader (fd_dup, make_function (
176- [py_fut, nbytes, fd=fd_dup] (object obj) {
157+ _async_wait_fd (fd_dup,
158+ [py_fut, nbytes, fd=fd_dup] {
177159 std::vector<char > buffer (nbytes);
178160 ssize_t nbytes_read = read (fd, buffer.data (), nbytes);
179161 py_fut.attr (" set_result" )(nbytes_read);
180- },
181- default_call_policies (), boost::mpl::vector< void , object>() ));
162+ },
163+ _read_key (fd ));
182164 return py_fut;
183165}
184166
@@ -189,12 +171,12 @@ object event_loop::sock_sendall(object sock, object data)
189171 char const * py_str = extract<char const *>(data.attr (" decode" )());
190172 ssize_t py_str_len = len (data);
191173 object py_fut = _pymod_concurrent_future.attr (" Future" )();
192- add_writer (fd_dup, make_function (
193- [py_fut, fd, py_str, py_str_len] (object obj) {
174+ _async_wait_fd (fd_dup,
175+ [py_fut, fd, py_str, py_str_len] {
194176 write (fd, py_str, py_str_len);
195177 py_fut.attr (" set_result" )(object ());
196- },
197- default_call_policies (), boost::mpl::vector< void , object>() ));
178+ },
179+ _write_key (fd ));
198180 return py_fut;
199181}
200182
@@ -205,22 +187,20 @@ object event_loop::sock_connect(object sock, object address)
205187 {
206188 // TODO: _ensure_resolve
207189 }
208- object fut = _pymod_concurrent_future.attr (" Future" )();
190+ object py_fut = _pymod_concurrent_future.attr (" Future" )();
209191 int fd = extract<int >(sock.attr (" fileno" )());
210192 try
211193 {
212194 sock.attr (" connect" )(address);
213- fut .attr (" set_result" )(object ());
195+ py_fut .attr (" set_result" )(object ());
214196 }
215197 catch (const error_already_set& e)
216198 {
217199 if (PyErr_ExceptionMatches (PyExc_BlockingIOError)
218200 || PyErr_ExceptionMatches (PyExc_InterruptedError))
219201 {
220202 PyErr_Clear ();
221- add_writer (dup (fd), make_function (bind (
222- _sock_connect_cb, _pymod_socket, fut, sock, address),
223- default_call_policies (), boost::mpl::vector<void , object>()));
203+ _async_wait_fd (dup (fd), bind (_sock_connect_cb, _pymod_socket, py_fut, sock, address), _write_key (fd));
224204 }
225205 else if (PyErr_ExceptionMatches (PyExc_SystemExit)
226206 || PyErr_ExceptionMatches (PyExc_KeyboardInterrupt))
@@ -230,17 +210,17 @@ object event_loop::sock_connect(object sock, object address)
230210 else
231211 {
232212 PyErr_Clear ();
233- fut .attr (" set_exception" )(std::current_exception ());
213+ py_fut .attr (" set_exception" )(std::current_exception ());
234214 }
235215 }
236- return fut ;
216+ return py_fut ;
237217}
238218
239219object event_loop::sock_accept (object sock)
240220{
241- object fut = _pymod_concurrent_future.attr (" Future" )();
242- _sock_accept (*this , fut , sock);
243- return fut ;
221+ object py_fut = _pymod_concurrent_future.attr (" Future" )();
222+ _sock_accept (*this , py_fut , sock);
223+ return py_fut ;
244224}
245225
246226// TODO: implement this
@@ -263,26 +243,22 @@ object event_loop::start_tls(object transport, object protocol, object sslcontex
263243object event_loop::getaddrinfo (object host, int port, int family, int type, int proto, int flags)
264244{
265245 object py_fut = _pymod_concurrent_future.attr (" Future" )();
266- call_soon ( make_function (
267- [this , py_fut, host, port, family, type, proto, flags] (object obj) {
246+ _strand. post (
247+ [this , py_fut, host, port, family, type, proto, flags] {
268248 object res = _pymod_socket.attr (" getaddrinfo" )(host, port, family, type, proto, flags);
269249 py_fut.attr (" set_result" )(res);
270- },
271- default_call_policies (),
272- boost::mpl::vector<void , object>()));
250+ });
273251 return py_fut;
274252}
275253
276254object event_loop::getnameinfo (object sockaddr, int flags)
277255{
278256 object py_fut = _pymod_concurrent_future.attr (" Future" )();
279- call_soon ( make_function (
280- [this , py_fut, sockaddr, flags] (object obj) {
257+ _strand. post (
258+ [this , py_fut, sockaddr, flags] {
281259 object res = _pymod_socket.attr (" getnameinfo" )(sockaddr, flags);
282260 py_fut.attr (" set_result" )(res);
283- },
284- default_call_policies (),
285- boost::mpl::vector<void , object>()));
261+ });
286262 return py_fut;
287263}
288264
0 commit comments