blob: 65399538fdc8a4c702321431b5866fc9e91a5111 [file] [log] [blame]
Austin Schuhe84c3ed2019-12-14 15:29:48 -08001#include "aos/network/sctp_lib.h"
2
3#include <arpa/inet.h>
4#include <net/if.h>
5#include <netdb.h>
6#include <netinet/sctp.h>
Austin Schuh2fe4b712020-03-15 14:21:45 -07007#include <sys/stat.h>
8#include <sys/types.h>
9#include <unistd.h>
Austin Schuhe84c3ed2019-12-14 15:29:48 -080010
Austin Schuha705d782021-07-31 20:40:00 -070011#include <algorithm>
Austin Schuhe84c3ed2019-12-14 15:29:48 -080012#include <string_view>
13
Austin Schuh2fe4b712020-03-15 14:21:45 -070014#include "aos/util/file.h"
15
Austin Schuh0a0a8272021-12-08 13:19:32 -080016DEFINE_string(interface, "", "network interface");
Brian Silverman0c6d44e2021-11-10 12:27:49 -080017DEFINE_bool(disable_ipv6, false, "disable ipv6");
Austin Schuh9dd8f592021-12-25 14:32:43 -080018DEFINE_int32(rmem, 0, "If nonzero, set rmem to this size.");
Austin Schuhe84c3ed2019-12-14 15:29:48 -080019
20namespace aos {
21namespace message_bridge {
22
23namespace {
24const char *sac_state_tbl[] = {"COMMUNICATION_UP", "COMMUNICATION_LOST",
25 "RESTART", "SHUTDOWN_COMPLETE",
Sarah Newman4aeb2372022-04-06 13:07:11 -070026 "CANT_START_ASSOCIATION"};
Austin Schuhe84c3ed2019-12-14 15:29:48 -080027
28typedef union {
29 struct sctp_initmsg init;
30 struct sctp_sndrcvinfo sndrcvinfo;
31} _sctp_cmsg_data_t;
32
33} // namespace
34
Austin Schuh0a0a8272021-12-08 13:19:32 -080035bool Ipv6Enabled() {
36 if (FLAGS_disable_ipv6) {
37 return false;
38 }
39 int fd = socket(AF_INET6, SOCK_SEQPACKET, IPPROTO_SCTP);
40 if (fd != -1) {
41 close(fd);
42 return true;
43 }
44 switch (errno) {
45 case EAFNOSUPPORT:
46 case EINVAL:
47 case EPROTONOSUPPORT:
48 PLOG(INFO) << "no ipv6";
49 return false;
50 default:
51 PLOG(FATAL) << "Open socket failed";
52 return false;
53 };
54}
55
56struct sockaddr_storage ResolveSocket(std::string_view host, int port,
57 bool use_ipv6) {
Austin Schuhe84c3ed2019-12-14 15:29:48 -080058 struct sockaddr_storage result;
James Kuszmaul784deb72023-02-17 14:42:51 -080059 memset(&result, 0, sizeof(result));
Austin Schuhe84c3ed2019-12-14 15:29:48 -080060 struct addrinfo *addrinfo_result;
61 struct sockaddr_in *t_addr = (struct sockaddr_in *)&result;
62 struct sockaddr_in6 *t_addr6 = (struct sockaddr_in6 *)&result;
Brian Silverman0c6d44e2021-11-10 12:27:49 -080063 struct addrinfo hints;
64 memset(&hints, 0, sizeof(hints));
Austin Schuh0a0a8272021-12-08 13:19:32 -080065 if (!use_ipv6) {
Brian Silverman0c6d44e2021-11-10 12:27:49 -080066 hints.ai_family = AF_INET;
67 } else {
Austin Schuh0a0a8272021-12-08 13:19:32 -080068 // Default to IPv6 as the clearly superior protocol, since it also handles
69 // IPv4.
Brian Silverman0c6d44e2021-11-10 12:27:49 -080070 hints.ai_family = AF_INET6;
71 }
72 hints.ai_socktype = SOCK_SEQPACKET;
73 hints.ai_protocol = IPPROTO_SCTP;
74 // We deliberately avoid AI_ADDRCONFIG here because it breaks running things
75 // inside Bazel's test sandbox, which has no non-localhost IPv4 or IPv6
76 // addresses. Also, it's not really helpful, because most systems will have
77 // link-local addresses of both types with any interface that's up.
78 hints.ai_flags = AI_PASSIVE | AI_V4MAPPED | AI_NUMERICSERV;
79 int ret = getaddrinfo(host.empty() ? nullptr : std::string(host).c_str(),
80 std::to_string(port).c_str(), &hints, &addrinfo_result);
Brian Silverman0c6d44e2021-11-10 12:27:49 -080081 if (ret == EAI_SYSTEM) {
82 PLOG(FATAL) << "getaddrinfo failed to look up '" << host << "'";
83 } else if (ret != 0) {
84 LOG(FATAL) << "getaddrinfo failed to look up '" << host
85 << "': " << gai_strerror(ret);
86 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -080087 switch (addrinfo_result->ai_family) {
88 case AF_INET:
89 memcpy(t_addr, addrinfo_result->ai_addr, addrinfo_result->ai_addrlen);
90 t_addr->sin_family = addrinfo_result->ai_family;
91 t_addr->sin_port = htons(port);
92
93 break;
94 case AF_INET6:
95 memcpy(t_addr6, addrinfo_result->ai_addr, addrinfo_result->ai_addrlen);
96 t_addr6->sin6_family = addrinfo_result->ai_family;
97 t_addr6->sin6_port = htons(port);
98
99 if (FLAGS_interface.size() > 0) {
100 t_addr6->sin6_scope_id = if_nametoindex(FLAGS_interface.c_str());
101 }
102
103 break;
104 }
105
106 // Now print it back out nicely.
107 char host_string[NI_MAXHOST];
108 char service_string[NI_MAXSERV];
109
110 int error = getnameinfo((struct sockaddr *)&result,
111 addrinfo_result->ai_addrlen, host_string, NI_MAXHOST,
112 service_string, NI_MAXSERV, NI_NUMERICHOST);
113
114 if (error) {
115 LOG(ERROR) << "Reverse lookup failed ... " << gai_strerror(error);
116 }
117
118 LOG(INFO) << "remote:addr=" << host_string << ", port=" << service_string
119 << ", family=" << addrinfo_result->ai_family;
120
121 freeaddrinfo(addrinfo_result);
122
123 return result;
124}
125
126std::string_view Family(const struct sockaddr_storage &sockaddr) {
127 if (sockaddr.ss_family == AF_INET) {
128 return "AF_INET";
129 } else if (sockaddr.ss_family == AF_INET6) {
130 return "AF_INET6";
131 } else {
132 return "unknown";
133 }
134}
135std::string Address(const struct sockaddr_storage &sockaddr) {
136 char addrbuf[INET6_ADDRSTRLEN];
137 if (sockaddr.ss_family == AF_INET) {
138 const struct sockaddr_in *sin = (const struct sockaddr_in *)&sockaddr;
139 return std::string(
140 inet_ntop(AF_INET, &sin->sin_addr, addrbuf, INET6_ADDRSTRLEN));
141 } else {
142 const struct sockaddr_in6 *sin6 = (const struct sockaddr_in6 *)&sockaddr;
143 return std::string(
144 inet_ntop(AF_INET6, &sin6->sin6_addr, addrbuf, INET6_ADDRSTRLEN));
145 }
146}
147
148void PrintNotification(const Message *msg) {
149 const union sctp_notification *snp =
150 (const union sctp_notification *)msg->data();
151
152 LOG(INFO) << "Notification:";
153
154 switch (snp->sn_header.sn_type) {
155 case SCTP_ASSOC_CHANGE: {
156 const struct sctp_assoc_change *sac = &snp->sn_assoc_change;
157 LOG(INFO) << "SCTP_ASSOC_CHANGE(" << sac_state_tbl[sac->sac_state] << ")";
158 VLOG(1) << " (assoc_change: state=" << sac->sac_state
159 << ", error=" << sac->sac_error
160 << ", instr=" << sac->sac_inbound_streams
161 << " outstr=" << sac->sac_outbound_streams
162 << ", assoc=" << sac->sac_assoc_id << ")";
163 } break;
164 case SCTP_PEER_ADDR_CHANGE: {
165 const struct sctp_paddr_change *spc = &snp->sn_paddr_change;
166 LOG(INFO) << " SlCTP_PEER_ADDR_CHANGE";
167 VLOG(1) << "\t\t(peer_addr_change: " << Address(spc->spc_aaddr)
168 << " state=" << spc->spc_state << ", error=" << spc->spc_error
169 << ")";
170 } break;
171 case SCTP_SEND_FAILED: {
172 const struct sctp_send_failed *ssf = &snp->sn_send_failed;
173 LOG(INFO) << " SCTP_SEND_FAILED";
174 VLOG(1) << "\t\t(sendfailed: len=" << ssf->ssf_length
175 << " err=" << ssf->ssf_error << ")";
176 } break;
177 case SCTP_REMOTE_ERROR: {
178 const struct sctp_remote_error *sre = &snp->sn_remote_error;
179 LOG(INFO) << " SCTP_REMOTE_ERROR";
180 VLOG(1) << "\t\t(remote_error: err=" << ntohs(sre->sre_error) << ")";
181 } break;
Austin Schuhf7777002020-09-01 18:41:28 -0700182 case SCTP_STREAM_CHANGE_EVENT: {
Austin Schuh62a0c272021-03-31 21:04:53 -0700183 const struct sctp_stream_change_event *sce = &snp->sn_strchange_event;
Austin Schuhf7777002020-09-01 18:41:28 -0700184 LOG(INFO) << " SCTP_STREAM_CHANGE_EVENT";
185 VLOG(1) << "\t\t(stream_change_event: flags=" << sce->strchange_flags
186 << ", assoc_id=" << sce->strchange_assoc_id
187 << ", instrms=" << sce->strchange_instrms
188 << ", outstrms=" << sce->strchange_outstrms << " )";
189 } break;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800190 case SCTP_SHUTDOWN_EVENT: {
191 LOG(INFO) << " SCTP_SHUTDOWN_EVENT";
192 } break;
193 default:
194 LOG(INFO) << " Unknown type: " << snp->sn_header.sn_type;
195 break;
196 }
197}
198
199std::string GetHostname() {
200 char buf[256];
201 buf[sizeof(buf) - 1] = '\0';
202 PCHECK(gethostname(buf, sizeof(buf) - 1) == 0);
203 return buf;
204}
205
206std::string Message::PeerAddress() const { return Address(sin); }
207
208void LogSctpStatus(int fd, sctp_assoc_t assoc_id) {
209 struct sctp_status status;
210 memset(&status, 0, sizeof(status));
211 status.sstat_assoc_id = assoc_id;
212
213 socklen_t size = sizeof(status);
Austin Schuha5f545b2021-07-31 20:39:42 -0700214 const int result = getsockopt(fd, SOL_SCTP, SCTP_STATUS,
215 reinterpret_cast<void *>(&status), &size);
216 if (result == -1 && errno == EINVAL) {
217 LOG(INFO) << "sctp_status) not associated";
218 return;
219 }
220 PCHECK(result == 0);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800221
222 LOG(INFO) << "sctp_status) sstat_assoc_id:" << status.sstat_assoc_id
223 << " sstat_state:" << status.sstat_state
224 << " sstat_rwnd:" << status.sstat_rwnd
225 << " sstat_unackdata:" << status.sstat_unackdata
226 << " sstat_penddata:" << status.sstat_penddata
227 << " sstat_instrms:" << status.sstat_instrms
228 << " sstat_outstrms:" << status.sstat_outstrms
229 << " sstat_fragmentation_point:" << status.sstat_fragmentation_point
230 << " sstat_primary.spinfo_srtt:" << status.sstat_primary.spinfo_srtt
231 << " sstat_primary.spinfo_rto:" << status.sstat_primary.spinfo_rto;
232}
233
Austin Schuh507f7582021-07-31 20:39:55 -0700234void SctpReadWrite::OpenSocket(const struct sockaddr_storage &sockaddr_local) {
235 fd_ = socket(sockaddr_local.ss_family, SOCK_SEQPACKET, IPPROTO_SCTP);
236 PCHECK(fd_ != -1);
237 LOG(INFO) << "socket(" << Family(sockaddr_local)
238 << ", SOCK_SEQPACKET, IPPROTOSCTP) = " << fd_;
239 {
240 // Per https://tools.ietf.org/html/rfc6458
241 // Setting this to !0 allows event notifications to be interleaved
Austin Schuha705d782021-07-31 20:40:00 -0700242 // with data if enabled. This typically only matters during congestion.
243 // However, Linux seems to interleave under memory pressure regardless of
244 // this being enabled, so we have to handle it in the code anyways, so might
245 // as well turn it on all the time.
246 // TODO(Brian): Change this to 2 once we have kernels that support it, and
247 // also address the TODO in ProcessNotification to match on all the
248 // necessary fields.
249 int interleaving = 1;
Austin Schuh507f7582021-07-31 20:39:55 -0700250 PCHECK(setsockopt(fd_, IPPROTO_SCTP, SCTP_FRAGMENT_INTERLEAVE,
251 &interleaving, sizeof(interleaving)) == 0);
252 }
253 {
254 // Enable recvinfo when a packet arrives.
255 int on = 1;
256 PCHECK(setsockopt(fd_, IPPROTO_SCTP, SCTP_RECVRCVINFO, &on, sizeof(int)) ==
257 0);
258 }
259
Austin Schuha705d782021-07-31 20:40:00 -0700260 {
261 // TODO(austin): This is the old style registration... But, the sctp
262 // stack out in the wild for linux is old and primitive.
263 struct sctp_event_subscribe subscribe;
264 memset(&subscribe, 0, sizeof(subscribe));
265 subscribe.sctp_association_event = 1;
266 subscribe.sctp_stream_change_event = 1;
267 subscribe.sctp_partial_delivery_event = 1;
268 PCHECK(setsockopt(fd(), SOL_SCTP, SCTP_EVENTS, (char *)&subscribe,
269 sizeof(subscribe)) == 0);
270 }
271
Austin Schuh507f7582021-07-31 20:39:55 -0700272 DoSetMaxSize();
273}
274
275bool SctpReadWrite::SendMessage(
276 int stream, std::string_view data, int time_to_live,
277 std::optional<struct sockaddr_storage> sockaddr_remote,
278 sctp_assoc_t snd_assoc_id) {
279 CHECK(fd_ != -1);
280 struct iovec iov;
281 iov.iov_base = const_cast<char *>(data.data());
282 iov.iov_len = data.size();
283
284 // Use the assoc_id for the destination instead of the msg_name.
285 struct msghdr outmsg;
James Kuszmaul784deb72023-02-17 14:42:51 -0800286 memset(&outmsg, 0, sizeof(outmsg));
Austin Schuh507f7582021-07-31 20:39:55 -0700287 if (sockaddr_remote) {
288 outmsg.msg_name = &*sockaddr_remote;
289 outmsg.msg_namelen = sizeof(*sockaddr_remote);
Sarah Newman4aeb2372022-04-06 13:07:11 -0700290 VLOG(2) << "Sending to " << Address(*sockaddr_remote);
Austin Schuh507f7582021-07-31 20:39:55 -0700291 } else {
292 outmsg.msg_namelen = 0;
293 }
294
295 // Data to send.
296 outmsg.msg_iov = &iov;
297 outmsg.msg_iovlen = 1;
298
299 // Build up the sndinfo message.
300 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
301 outmsg.msg_control = outcmsg;
302 outmsg.msg_controllen = sizeof(outcmsg);
303 outmsg.msg_flags = 0;
304
305 struct cmsghdr *cmsg = CMSG_FIRSTHDR(&outmsg);
306 cmsg->cmsg_level = IPPROTO_SCTP;
307 cmsg->cmsg_type = SCTP_SNDRCV;
308 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
309
310 struct sctp_sndrcvinfo *sinfo =
311 reinterpret_cast<struct sctp_sndrcvinfo *>(CMSG_DATA(cmsg));
312 memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
313 sinfo->sinfo_ppid = ++send_ppid_;
314 sinfo->sinfo_stream = stream;
315 sinfo->sinfo_flags = 0;
316 sinfo->sinfo_assoc_id = snd_assoc_id;
317 sinfo->sinfo_timetolive = time_to_live;
318
319 // And send.
320 const ssize_t size = sendmsg(fd_, &outmsg, MSG_NOSIGNAL | MSG_DONTWAIT);
321 if (size == -1) {
322 if (errno == EPIPE || errno == EAGAIN || errno == ESHUTDOWN ||
323 errno == EINTR) {
Austin Schuh581fab92022-02-07 19:50:54 -0800324 if (VLOG_IS_ON(1)) {
325 PLOG(WARNING) << "sendmsg on sctp socket failed";
326 }
Austin Schuh507f7582021-07-31 20:39:55 -0700327 return false;
328 }
329 PLOG(FATAL) << "sendmsg on sctp socket failed";
330 return false;
331 }
332 CHECK_EQ(static_cast<ssize_t>(data.size()), size);
Sarah Newman4aeb2372022-04-06 13:07:11 -0700333 VLOG(2) << "Sent " << data.size();
Austin Schuh507f7582021-07-31 20:39:55 -0700334 return true;
335}
336
Austin Schuha705d782021-07-31 20:40:00 -0700337// We read each fragment into a fresh Message, because most of them won't be
338// fragmented. If we do end up with a fragment, then we copy the data out of it.
Austin Schuh507f7582021-07-31 20:39:55 -0700339aos::unique_c_ptr<Message> SctpReadWrite::ReadMessage() {
340 CHECK(fd_ != -1);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800341
Austin Schuha705d782021-07-31 20:40:00 -0700342 while (true) {
343 aos::unique_c_ptr<Message> result(
344 reinterpret_cast<Message *>(malloc(sizeof(Message) + max_size_ + 1)));
345
Austin Schuh05c18122021-07-31 20:39:47 -0700346 struct msghdr inmessage;
Austin Schuhc4202572021-03-31 21:06:55 -0700347 memset(&inmessage, 0, sizeof(struct msghdr));
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800348
Austin Schuh05c18122021-07-31 20:39:47 -0700349 struct iovec iov;
Austin Schuha705d782021-07-31 20:40:00 -0700350 iov.iov_len = max_size_ + 1;
351 iov.iov_base = result->mutable_data();
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800352
Austin Schuhc4202572021-03-31 21:06:55 -0700353 inmessage.msg_iov = &iov;
354 inmessage.msg_iovlen = 1;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800355
Austin Schuh05c18122021-07-31 20:39:47 -0700356 char incmsg[CMSG_SPACE(sizeof(_sctp_cmsg_data_t))];
Austin Schuhc4202572021-03-31 21:06:55 -0700357 inmessage.msg_control = incmsg;
358 inmessage.msg_controllen = sizeof(incmsg);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800359
Austin Schuhc4202572021-03-31 21:06:55 -0700360 inmessage.msg_namelen = sizeof(struct sockaddr_storage);
361 inmessage.msg_name = &result->sin;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800362
Austin Schuha705d782021-07-31 20:40:00 -0700363 const ssize_t size = recvmsg(fd_, &inmessage, MSG_DONTWAIT);
364 if (size == -1) {
365 if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
366 // These are all non-fatal failures indicating we should retry later.
367 return nullptr;
Austin Schuhc4202572021-03-31 21:06:55 -0700368 }
Austin Schuha705d782021-07-31 20:40:00 -0700369 PLOG(FATAL) << "recvmsg on sctp socket " << fd_ << " failed";
Austin Schuhc4202572021-03-31 21:06:55 -0700370 }
371
Austin Schuha705d782021-07-31 20:40:00 -0700372 CHECK(!(inmessage.msg_flags & MSG_CTRUNC))
Austin Schuhc4202572021-03-31 21:06:55 -0700373 << ": Control message truncated.";
374
Austin Schuha705d782021-07-31 20:40:00 -0700375 CHECK_LE(size, static_cast<ssize_t>(max_size_))
Austin Schuh507f7582021-07-31 20:39:55 -0700376 << ": Message overflowed buffer on stream "
377 << result->header.rcvinfo.rcv_sid << ".";
Austin Schuhc4202572021-03-31 21:06:55 -0700378
Austin Schuha705d782021-07-31 20:40:00 -0700379 result->size = size;
380 if (MSG_NOTIFICATION & inmessage.msg_flags) {
381 result->message_type = Message::kNotification;
382 } else {
383 result->message_type = Message::kMessage;
384 }
385 result->partial_deliveries = 0;
Austin Schuhc4202572021-03-31 21:06:55 -0700386
Austin Schuha705d782021-07-31 20:40:00 -0700387 {
388 bool found_rcvinfo = false;
389 for (struct cmsghdr *scmsg = CMSG_FIRSTHDR(&inmessage); scmsg != NULL;
390 scmsg = CMSG_NXTHDR(&inmessage, scmsg)) {
391 switch (scmsg->cmsg_type) {
392 case SCTP_RCVINFO: {
393 CHECK(!found_rcvinfo);
394 found_rcvinfo = true;
395 result->header.rcvinfo =
396 *reinterpret_cast<struct sctp_rcvinfo *>(CMSG_DATA(scmsg));
397 } break;
398 default:
399 LOG(INFO) << "\tUnknown type: " << scmsg->cmsg_type;
400 break;
401 }
402 }
403 CHECK_EQ(found_rcvinfo, result->message_type == Message::kMessage)
404 << ": Failed to find a SCTP_RCVINFO cmsghdr. flags: "
405 << inmessage.msg_flags;
406 }
407 if (result->message_type == Message::kNotification) {
408 // Notifications are never fragmented, just return it now.
409 CHECK(inmessage.msg_flags & MSG_EOR)
410 << ": Notifications should never be big enough to fragment";
411 if (ProcessNotification(result.get())) {
412 // We handled this notification internally, so don't pass it on.
413 return nullptr;
414 }
415 return result;
416 }
417
418 auto partial_message_iterator =
419 std::find_if(partial_messages_.begin(), partial_messages_.end(),
420 [&result](const aos::unique_c_ptr<Message> &candidate) {
421 return result->header.rcvinfo.rcv_sid ==
422 candidate->header.rcvinfo.rcv_sid &&
423 result->header.rcvinfo.rcv_ssn ==
424 candidate->header.rcvinfo.rcv_ssn &&
425 result->header.rcvinfo.rcv_assoc_id ==
426 candidate->header.rcvinfo.rcv_assoc_id;
427 });
428 if (partial_message_iterator != partial_messages_.end()) {
429 const aos::unique_c_ptr<Message> &partial_message =
430 *partial_message_iterator;
431 // Verify it's really part of the same message.
432 CHECK_EQ(partial_message->message_type, result->message_type)
433 << ": for " << result->header.rcvinfo.rcv_sid << ","
434 << result->header.rcvinfo.rcv_ssn << ","
435 << result->header.rcvinfo.rcv_assoc_id;
436 CHECK_EQ(partial_message->header.rcvinfo.rcv_ppid,
437 result->header.rcvinfo.rcv_ppid)
438 << ": for " << result->header.rcvinfo.rcv_sid << ","
439 << result->header.rcvinfo.rcv_ssn << ","
440 << result->header.rcvinfo.rcv_assoc_id;
441
442 // Now copy the data over and update the size.
443 CHECK_LE(partial_message->size + result->size, max_size_)
444 << ": Assembled fragments overflowed buffer on stream "
445 << result->header.rcvinfo.rcv_sid << ".";
446 memcpy(partial_message->mutable_data() + partial_message->size,
447 result->data(), result->size);
448 ++partial_message->partial_deliveries;
Sarah Newman4aeb2372022-04-06 13:07:11 -0700449 VLOG(2) << "Merged fragment of " << result->size << " after "
Austin Schuha705d782021-07-31 20:40:00 -0700450 << partial_message->size << ", had "
451 << partial_message->partial_deliveries
452 << ", for: " << result->header.rcvinfo.rcv_sid << ","
453 << result->header.rcvinfo.rcv_ssn << ","
454 << result->header.rcvinfo.rcv_assoc_id;
455 partial_message->size += result->size;
456 result.reset();
457 }
458
459 if (inmessage.msg_flags & MSG_EOR) {
460 // This is the last fragment, so we have something to return.
461 if (partial_message_iterator != partial_messages_.end()) {
462 // It was already merged into the message in the list, so now we pull
463 // that out of the list and return it.
464 CHECK(!result);
465 result = std::move(*partial_message_iterator);
466 partial_messages_.erase(partial_message_iterator);
467 VLOG(1) << "Final count: " << (result->partial_deliveries + 1)
468 << ", size: " << result->size
469 << ", for: " << result->header.rcvinfo.rcv_sid << ","
470 << result->header.rcvinfo.rcv_ssn << ","
471 << result->header.rcvinfo.rcv_assoc_id;
472 }
473 CHECK(result);
474 return result;
475 }
476 if (partial_message_iterator == partial_messages_.end()) {
Sarah Newman4aeb2372022-04-06 13:07:11 -0700477 VLOG(2) << "Starting fragment for: " << result->header.rcvinfo.rcv_sid
Austin Schuha705d782021-07-31 20:40:00 -0700478 << "," << result->header.rcvinfo.rcv_ssn << ","
479 << result->header.rcvinfo.rcv_assoc_id;
480 // Need to record this as the first fragment.
481 partial_messages_.emplace_back(std::move(result));
482 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800483 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800484}
485
Sarah Newman80e955e2022-04-13 11:19:36 -0700486bool SctpReadWrite::Abort(sctp_assoc_t snd_assoc_id) {
487 if (fd_ == -1) {
488 return true;
489 }
490 VLOG(1) << "Sending abort to assoc " << snd_assoc_id;
491
492 // Use the assoc_id for the destination instead of the msg_name.
493 struct msghdr outmsg;
James Kuszmaul784deb72023-02-17 14:42:51 -0800494 memset(&outmsg, 0, sizeof(outmsg));
Sarah Newman80e955e2022-04-13 11:19:36 -0700495 outmsg.msg_namelen = 0;
496
497 outmsg.msg_iovlen = 0;
498
499 // Build up the sndinfo message.
500 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
501 outmsg.msg_control = outcmsg;
502 outmsg.msg_controllen = CMSG_SPACE(sizeof(struct sctp_sndrcvinfo));
503 outmsg.msg_flags = 0;
504
505 struct cmsghdr *cmsg = CMSG_FIRSTHDR(&outmsg);
506 cmsg->cmsg_level = IPPROTO_SCTP;
507 cmsg->cmsg_type = SCTP_SNDRCV;
508 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
509
510 struct sctp_sndrcvinfo *sinfo = (struct sctp_sndrcvinfo *)CMSG_DATA(cmsg);
511 memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
512 sinfo->sinfo_stream = 0;
513 sinfo->sinfo_flags = SCTP_ABORT;
514 sinfo->sinfo_assoc_id = snd_assoc_id;
515
516 // And send.
517 const ssize_t size = sendmsg(fd_, &outmsg, MSG_NOSIGNAL | MSG_DONTWAIT);
518 if (size == -1) {
519 if (errno == EPIPE || errno == EAGAIN || errno == ESHUTDOWN) {
520 return false;
521 }
522 return false;
523 } else {
524 CHECK_EQ(0, size);
525 return true;
526 }
527}
528
Austin Schuh507f7582021-07-31 20:39:55 -0700529void SctpReadWrite::CloseSocket() {
530 if (fd_ == -1) {
531 return;
532 }
533 LOG(INFO) << "close(" << fd_ << ")";
534 PCHECK(close(fd_) == 0);
535 fd_ = -1;
536}
537
538void SctpReadWrite::DoSetMaxSize() {
Austin Schuh61224882021-10-11 18:21:11 -0700539 size_t max_size = max_size_;
Austin Schuh507f7582021-07-31 20:39:55 -0700540
Austin Schuh61224882021-10-11 18:21:11 -0700541 // This sets the max packet size that we can send.
Austin Schuh507f7582021-07-31 20:39:55 -0700542 CHECK_GE(ReadWMemMax(), max_size)
543 << "wmem_max is too low. To increase wmem_max temporarily, do sysctl "
544 "-w net.core.wmem_max="
545 << max_size;
Austin Schuh507f7582021-07-31 20:39:55 -0700546 PCHECK(setsockopt(fd(), SOL_SOCKET, SO_SNDBUF, &max_size, sizeof(max_size)) ==
547 0);
Austin Schuh61224882021-10-11 18:21:11 -0700548
549 // The SO_RCVBUF option (also controlled by net.core.rmem_default) needs to be
550 // decently large but the actual size can be measured by tuning. The defaults
551 // should be fine. If it isn't big enough, transmission will fail.
Austin Schuh9dd8f592021-12-25 14:32:43 -0800552 if (FLAGS_rmem > 0) {
553 size_t rmem = FLAGS_rmem;
554 PCHECK(setsockopt(fd(), SOL_SOCKET, SO_RCVBUF, &rmem, sizeof(rmem)) == 0);
555 }
Austin Schuh507f7582021-07-31 20:39:55 -0700556}
557
Austin Schuha705d782021-07-31 20:40:00 -0700558bool SctpReadWrite::ProcessNotification(const Message *message) {
559 const union sctp_notification *const snp =
560 reinterpret_cast<const union sctp_notification *>(message->data());
561 switch (snp->sn_header.sn_type) {
562 case SCTP_PARTIAL_DELIVERY_EVENT: {
563 const struct sctp_pdapi_event *const partial_delivery =
564 &snp->sn_pdapi_event;
565 CHECK_EQ(partial_delivery->pdapi_length, sizeof(*partial_delivery))
566 << ": Kernel's SCTP code is not a version we support";
567 switch (partial_delivery->pdapi_indication) {
568 case SCTP_PARTIAL_DELIVERY_ABORTED: {
569 const auto iterator = std::find_if(
570 partial_messages_.begin(), partial_messages_.end(),
571 [partial_delivery](const aos::unique_c_ptr<Message> &candidate) {
572 // TODO(Brian): Once we have new enough userpace headers, for
573 // kernels that support level-2 interleaving, we'll need to add
574 // this:
575 // candidate->header.rcvinfo.rcv_sid ==
576 // partial_delivery->pdapi_stream &&
577 // candidate->header.rcvinfo.rcv_ssn ==
578 // partial_delivery->pdapi_seq &&
579 return candidate->header.rcvinfo.rcv_assoc_id ==
580 partial_delivery->pdapi_assoc_id;
581 });
582 CHECK(iterator != partial_messages_.end())
583 << ": Got out of sync with the kernel for "
584 << partial_delivery->pdapi_assoc_id;
585 VLOG(1) << "Pruning partial delivery for "
586 << iterator->get()->header.rcvinfo.rcv_sid << ","
587 << iterator->get()->header.rcvinfo.rcv_ssn << ","
588 << iterator->get()->header.rcvinfo.rcv_assoc_id;
589 partial_messages_.erase(iterator);
590 }
591 return true;
592 }
593 } break;
594 }
595 return false;
596}
597
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800598void Message::LogRcvInfo() const {
599 LOG(INFO) << "\tSNDRCV (stream=" << header.rcvinfo.rcv_sid
600 << " ssn=" << header.rcvinfo.rcv_ssn
601 << " tsn=" << header.rcvinfo.rcv_tsn << " flags=0x" << std::hex
602 << header.rcvinfo.rcv_flags << std::dec
603 << " ppid=" << header.rcvinfo.rcv_ppid
604 << " cumtsn=" << header.rcvinfo.rcv_cumtsn << ")";
605}
606
Austin Schuh2fe4b712020-03-15 14:21:45 -0700607size_t ReadRMemMax() {
608 struct stat current_stat;
609 if (stat("/proc/sys/net/core/rmem_max", &current_stat) != -1) {
610 return static_cast<size_t>(
611 std::stoi(util::ReadFileToStringOrDie("/proc/sys/net/core/rmem_max")));
612 } else {
613 LOG(WARNING) << "/proc/sys/net/core/rmem_max doesn't exist. Are you in a "
614 "container?";
615 return 212992;
616 }
617}
618
619size_t ReadWMemMax() {
620 struct stat current_stat;
621 if (stat("/proc/sys/net/core/wmem_max", &current_stat) != -1) {
622 return static_cast<size_t>(
623 std::stoi(util::ReadFileToStringOrDie("/proc/sys/net/core/wmem_max")));
624 } else {
625 LOG(WARNING) << "/proc/sys/net/core/wmem_max doesn't exist. Are you in a "
626 "container?";
627 return 212992;
628 }
629}
630
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800631} // namespace message_bridge
632} // namespace aos