Line data Source code
1 : #include <array>
2 : #include <iostream>
3 : #include "socket.hpp"
4 : #if !defined(BOOST_ASIO_WINDOWS)
5 : #include <poll.h>
6 : #include <sys/socket.h>
7 : #include <sys/uio.h>
8 : #endif
9 : #if defined(__linux__)
10 : #include <linux/sockios.h>
11 : #include <sys/ioctl.h>
12 : #endif
13 :
14 : namespace thinger::asio {
15 :
16 : std::atomic<unsigned long> socket::connections(0);
17 :
18 57796 : socket::socket(const std::string&, boost::asio::io_context& io_context)
19 57796 : : io_context_(io_context) {
20 57796 : ++connections;
21 57796 : }
22 :
23 57796 : socket::~socket() {
24 57796 : --connections;
25 57796 : }
26 :
27 262992 : boost::asio::io_context& socket::get_io_context() const {
28 262992 : return io_context_;
29 : }
30 :
31 15894 : bool socket::requires_handshake() const {
32 15894 : return false;
33 : }
34 :
35 0 : awaitable<boost::system::error_code> socket::handshake(const std::string& host) {
36 : co_return boost::system::error_code{};
37 0 : }
38 :
39 6632 : size_t socket::write_now(const std::vector<boost::asio::const_buffer>&, boost::system::error_code& ec) {
40 6632 : ec = boost::asio::error::operation_not_supported;
41 6632 : return 0;
42 : }
43 :
44 32485 : size_t socket::send_now(native_handle_type descriptor, const std::vector<boost::asio::const_buffer>& buffers,
45 : boost::system::error_code& ec) {
46 : #if defined(MSG_DONTWAIT) && !defined(BOOST_ASIO_WINDOWS)
47 : std::array<iovec, 64> iov;
48 32485 : size_t count = std::min(buffers.size(), iov.size());
49 374544 : for (size_t i = 0; i < count; ++i) {
50 342059 : iov[i].iov_base = const_cast<void*>(buffers[i].data());
51 342059 : iov[i].iov_len = buffers[i].size();
52 : }
53 32485 : msghdr message{};
54 32485 : message.msg_iov = iov.data();
55 32485 : message.msg_iovlen = count;
56 32485 : int flags = MSG_DONTWAIT;
57 : #if defined(MSG_NOSIGNAL)
58 32485 : flags |= MSG_NOSIGNAL;
59 : #endif
60 : ssize_t bytes;
61 : do {
62 32485 : bytes = ::sendmsg(descriptor, &message, flags);
63 32485 : } while (bytes < 0 && errno == EINTR);
64 32485 : if (bytes < 0) {
65 344 : ec = boost::system::error_code(errno, boost::asio::error::get_system_category());
66 344 : if (ec == boost::asio::error::try_again) ec = boost::asio::error::would_block;
67 344 : return 0;
68 : }
69 32141 : ec = {};
70 32141 : return static_cast<size_t>(bytes);
71 : #else
72 : ec = boost::asio::error::operation_not_supported;
73 : return 0;
74 : #endif
75 : }
76 :
77 0 : awaitable<boost::system::error_code> socket::peer_closed() {
78 : co_return boost::asio::error::operation_not_supported;
79 0 : }
80 :
81 0 : awaitable<boost::system::error_code> socket::connect(const std::string&, const std::string&, std::chrono::seconds) {
82 : co_return boost::asio::error::operation_not_supported;
83 0 : }
84 :
85 3 : awaitable<boost::system::error_code> socket::wait(boost::asio::socket_base::wait_type) {
86 : co_return boost::asio::error::operation_not_supported;
87 6 : }
88 :
89 15 : bool socket::is_secure() const {
90 15 : return false;
91 : }
92 :
93 6 : size_t socket::available() const {
94 6 : return 0;
95 : }
96 :
97 0 : bool socket::readable_now() const {
98 0 : return false;
99 : }
100 :
101 0 : size_t socket::send_backlog() const {
102 0 : return 0;
103 : }
104 :
105 470 : size_t socket::output_queue(native_handle_type descriptor) {
106 470 : int bytes = 0;
107 : #if defined(__linux__)
108 : // TCP: not acknowledged, sent or not; Unix: not read
109 470 : if (::ioctl(descriptor, SIOCOUTQ, &bytes) != 0) return 0;
110 : #elif defined(SO_NWRITE)
111 : socklen_t size = sizeof(bytes);
112 : if (::getsockopt(descriptor, SOL_SOCKET, SO_NWRITE, &bytes, &size) != 0) return 0;
113 : #else
114 : (void) descriptor;
115 : #endif
116 470 : return bytes > 0 ? size_t(bytes) : 0;
117 : }
118 :
119 3459 : bool socket::poll_readable(native_handle_type descriptor) {
120 : #if !defined(BOOST_ASIO_WINDOWS)
121 3459 : pollfd fd{};
122 3459 : fd.fd = descriptor;
123 3459 : fd.events = POLLIN;
124 : int ready;
125 : do {
126 3459 : ready = ::poll(&fd, 1, 0);
127 3459 : } while (ready < 0 && errno == EINTR);
128 : // POLLIN, POLLHUP, POLLERR or POLLNVAL, or poll itself failing
129 3459 : return ready != 0;
130 : #else
131 : (void) descriptor;
132 : return false;
133 : #endif
134 : }
135 :
136 258 : boost::system::error_code socket::peek_closed(native_handle_type descriptor) {
137 : #if defined(MSG_DONTWAIT) && !defined(BOOST_ASIO_WINDOWS)
138 : char byte;
139 : ssize_t bytes;
140 : do {
141 258 : bytes = ::recv(descriptor, &byte, 1, MSG_PEEK | MSG_DONTWAIT);
142 258 : } while (bytes < 0 && errno == EINTR);
143 258 : if (bytes > 0) return {};
144 226 : if (bytes == 0) return boost::asio::error::eof;
145 40 : boost::system::error_code ec(errno, boost::asio::error::get_system_category());
146 : // Nothing to read yet: no conclusion, as with data
147 40 : if (ec == boost::asio::error::would_block || ec == boost::asio::error::try_again) return {};
148 34 : return ec;
149 : #else
150 : (void) descriptor;
151 : return boost::asio::error::operation_not_supported;
152 : #endif
153 : }
154 :
155 6 : std::string socket::get_remote_ip() const {
156 6 : return {};
157 : }
158 :
159 0 : std::string socket::get_local_port() const {
160 0 : return {};
161 : }
162 :
163 0 : std::string socket::get_remote_port() const {
164 0 : return {};
165 : }
166 :
167 15 : std::optional<tls_session> socket::tls() const {
168 15 : return std::nullopt;
169 : }
170 :
171 3 : awaitable<io_result> socket::read_some_cancellable(uint8_t buffer[], size_t max_size,
172 : boost::asio::cancellation_slot slot) {
173 : co_return co_await boost::asio::co_spawn(io_context_, read_some(buffer, max_size),
174 : boost::asio::bind_cancellation_slot(slot, boost::asio::use_awaitable));
175 6 : }
176 :
177 261 : awaitable<io_result> socket::read(uint8_t buffer[], size_t size) {
178 : size_t total = 0;
179 : while (total < size) {
180 : auto [ec, bytes] = co_await read_some(buffer + total, size - total);
181 : total += bytes;
182 : if (ec) co_return io_result{ec, total};
183 : // a socket that keeps the guarantees never returns 0 bytes here: do not spin
184 : if (bytes == 0) co_return io_result{boost::asio::error::eof, total};
185 : }
186 : co_return io_result{boost::system::error_code{}, total};
187 522 : }
188 :
189 15 : awaitable<io_result> socket::read(boost::asio::streambuf& buffer, size_t size) {
190 : size_t total = 0;
191 : while (total < size) {
192 : boost::asio::streambuf::mutable_buffers_type space;
193 : try {
194 : space = buffer.prepare(size - total);
195 : } catch (const std::length_error&) {
196 : co_return io_result{boost::asio::error::no_buffer_space, total};
197 : }
198 : auto [ec, bytes] = co_await read_some(static_cast<uint8_t*>(space.data()), space.size());
199 : buffer.commit(bytes);
200 : total += bytes;
201 : if (ec) co_return io_result{ec, total};
202 : if (bytes == 0) co_return io_result{boost::asio::error::eof, total};
203 : }
204 : co_return io_result{boost::system::error_code{}, total};
205 30 : }
206 :
207 42 : awaitable<io_result> socket::read_until(boost::asio::streambuf& buffer, std::string_view delim) {
208 : // Look for the delimiter in what the buffer holds, then in what is read, as
209 : // boost::asio::async_read_until does
210 : size_t search_from = 0;
211 : for (;;) {
212 : auto data = buffer.data();
213 : std::string_view held(static_cast<const char*>(data.data()), data.size());
214 : auto position = held.find(delim, search_from);
215 : if (position != std::string_view::npos) {
216 : co_return io_result{boost::system::error_code{}, position + delim.size()};
217 : }
218 : // the delimiter may start in the last bytes held
219 : search_from = held.size() >= delim.size() ? held.size() - delim.size() + 1 : 0;
220 : if (buffer.size() >= buffer.max_size()) {
221 : co_return io_result{boost::asio::error::not_found, 0};
222 : }
223 : auto read_size = std::min<size_t>(std::max<size_t>(512, buffer.capacity() - buffer.size()),
224 : std::min<size_t>(65536, buffer.max_size() - buffer.size()));
225 : auto space = buffer.prepare(read_size);
226 : auto [ec, bytes] = co_await read_some(static_cast<uint8_t*>(space.data()), space.size());
227 : buffer.commit(bytes);
228 : if (ec) co_return io_result{ec, 0};
229 : if (bytes == 0) co_return io_result{boost::asio::error::eof, 0};
230 : }
231 84 : }
232 :
233 6 : awaitable<io_result> socket::write(std::string_view str) {
234 : co_return co_await write(reinterpret_cast<const uint8_t*>(str.data()), str.size());
235 12 : }
236 :
237 45 : awaitable<io_result> socket::write(const std::vector<boost::asio::const_buffer>& buffers) {
238 : size_t total = 0;
239 : for (const auto& buffer : buffers) {
240 : if (buffer.size() == 0) continue;
241 : auto [ec, bytes] = co_await write(static_cast<const uint8_t*>(buffer.data()), buffer.size());
242 : total += bytes;
243 : if (ec) co_return io_result{ec, total};
244 : }
245 : co_return io_result{boost::system::error_code{}, total};
246 90 : }
247 :
248 7531 : awaitable<io_result> socket::write_slices(std::vector<boost::asio::const_buffer>& buffers,
249 : const sync_callback<void()>& progress) {
250 : std::vector<boost::asio::const_buffer> slice;
251 : size_t total = 0;
252 : consume(buffers, 0);
253 : while (!buffers.empty()) {
254 : slice.clear();
255 : for (size_t size = 0, i = 0; i < buffers.size() && size < WRITE_SLICE_SIZE; ++i) {
256 : slice.push_back(boost::asio::buffer(buffers[i], WRITE_SLICE_SIZE - size));
257 : size += slice.back().size();
258 : }
259 : auto [ec, bytes] = co_await write(slice);
260 : total += bytes;
261 : consume(buffers, bytes);
262 : if (ec) co_return io_result{ec, total};
263 : progress();
264 : }
265 : co_return io_result{boost::system::error_code{}, total};
266 15062 : }
267 :
268 54326 : void socket::consume(std::vector<boost::asio::const_buffer>& buffers, size_t size) {
269 54326 : auto remaining = buffers.begin();
270 79376 : while (remaining != buffers.end() && size >= remaining->size()) {
271 25050 : size -= remaining->size();
272 25050 : ++remaining;
273 : }
274 54326 : if (remaining != buffers.end()) *remaining += size;
275 54326 : buffers.erase(buffers.begin(), remaining);
276 54326 : }
277 :
278 : }
|
