summary |
shortlog |
log |
commit | commitdiff |
tree
raw |
patch |
inline | side by side (from parent 1:
907b42d)
In order to handle messages that are possibly larger than the socket
buffer size set by wmem_max and rmem_max /proc files, ensure that the
recv-side reads the data chunk-wise rather than hanging on a
MSG_WAITALL.
In addition to fixing this issue, chances are that it will also help
fixing hangs detected due to UNIX socket buffers filling up. The
MSG_WAITALL behavior in such situations might be unexpected.
Acked-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>
/* FIXME : use data_size for something ? */
ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
/* FIXME : use data_size for something ? */
ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
- sizeof(struct lttcomm_relayd_add_stream), MSG_WAITALL);
+ sizeof(struct lttcomm_relayd_add_stream), 0);
if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
ERR("Relay didn't receive valid add_stream struct size : %d", ret);
ret = -1;
if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
ERR("Relay didn't receive valid add_stream struct size : %d", ret);
ret = -1;
}
ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
}
ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
- sizeof(struct lttcomm_relayd_close_stream), MSG_WAITALL);
+ sizeof(struct lttcomm_relayd_close_stream), 0);
if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
ERR("Relay didn't receive valid add_stream struct size : %d", ret);
ret = -1;
if (ret < sizeof(struct lttcomm_relayd_close_stream)) {
ERR("Relay didn't receive valid add_stream struct size : %d", ret);
ret = -1;
}
memset(data_buffer, 0, data_size);
DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
}
memset(data_buffer, 0, data_size);
DBG2("Relay receiving metadata, waiting for %" PRIu64 " bytes", data_size);
- ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size,
- MSG_WAITALL);
+ ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
if (ret < 0 || ret != data_size) {
ret = -1;
ERR("Relay didn't receive the whole metadata");
if (ret < 0 || ret != data_size) {
ret = -1;
ERR("Relay didn't receive the whole metadata");
session->version_check_done = 1;
/* Get version from the other side. */
session->version_check_done = 1;
/* Get version from the other side. */
- ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL);
+ ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
if (ret < 0 || ret != sizeof(msg)) {
ret = -1;
ERR("Relay failed to receive the version values.");
if (ret < 0 || ret != sizeof(msg)) {
ret = -1;
ERR("Relay failed to receive the version values.");
- ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), MSG_WAITALL);
+ ret = cmd->sock->ops->recvmsg(cmd->sock, &msg, sizeof(msg), 0);
if (ret < sizeof(msg)) {
ERR("Relay didn't receive valid data_pending struct size : %d", ret);
ret = -1;
if (ret < sizeof(msg)) {
ERR("Relay didn't receive valid data_pending struct size : %d", ret);
ret = -1;
uint32_t data_size;
ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
uint32_t data_size;
ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
- sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL);
+ sizeof(struct lttcomm_relayd_data_hdr), 0);
if (ret <= 0) {
ERR("Connections seems to be closed");
ret = -1;
if (ret <= 0) {
ERR("Connections seems to be closed");
ret = -1;
DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
data_size, stream_id, net_seq_num);
DBG3("Receiving data of size %u for stream id %" PRIu64 " seqnum %" PRIu64,
data_size, stream_id, net_seq_num);
- ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
+ ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, 0);
if (ret <= 0) {
ret = -1;
goto end_unlock;
if (ret <= 0) {
ret = -1;
goto end_unlock;
if (relay_connection->type == RELAY_CONTROL) {
ret = relay_connection->sock->ops->recvmsg(
relay_connection->sock, &recv_hdr,
if (relay_connection->type == RELAY_CONTROL) {
ret = relay_connection->sock->ops->recvmsg(
relay_connection->sock, &recv_hdr,
- sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
+ sizeof(struct lttcomm_relayd_hdr), 0);
/* connection closed */
if (ret <= 0) {
relay_cleanup_poll_connection(&events, pollfd);
/* connection closed */
if (ret <= 0) {
relay_cleanup_poll_connection(&events, pollfd);
struct msghdr msg;
struct iovec iov[1];
ssize_t ret = -1;
struct msghdr msg;
struct iovec iov[1];
ssize_t ret = -1;
memset(&msg, 0, sizeof(msg));
memset(&msg, 0, sizeof(msg));
msg.msg_name = (struct sockaddr *) &sock->sockaddr.addr.sin;
msg.msg_namelen = sizeof(sock->sockaddr.addr.sin);
msg.msg_name = (struct sockaddr *) &sock->sockaddr.addr.sin;
msg.msg_namelen = sizeof(sock->sockaddr.addr.sin);
- if (flags == 0) {
- flags = MSG_WAITALL;
- }
-
+ len_last = iov[0].iov_len;
ret = recvmsg(sock->fd, &msg, flags);
ret = recvmsg(sock->fd, &msg, flags);
- } while (ret < 0 && errno == EINTR);
+ if (ret > 0) {
+ iov[0].iov_base += ret;
+ iov[0].iov_len -= ret;
+ assert(ret <= len_last);
+ }
+ } while ((ret > 0 && ret < len_last) || (ret < 0 && errno == EINTR));
if (ret < 0) {
PERROR("recvmsg inet");
if (ret < 0) {
PERROR("recvmsg inet");
+ } else if (ret > 0) {
+ ret = len;
+ /* Else ret = 0 meaning an orderly shutdown. */
struct msghdr msg;
struct iovec iov[1];
ssize_t ret = -1;
struct msghdr msg;
struct iovec iov[1];
ssize_t ret = -1;
memset(&msg, 0, sizeof(msg));
memset(&msg, 0, sizeof(msg));
msg.msg_name = (struct sockaddr *) &sock->sockaddr.addr.sin6;
msg.msg_namelen = sizeof(sock->sockaddr.addr.sin6);
msg.msg_name = (struct sockaddr *) &sock->sockaddr.addr.sin6;
msg.msg_namelen = sizeof(sock->sockaddr.addr.sin6);
- if (flags == 0) {
- flags = MSG_WAITALL;
- }
-
+ len_last = iov[0].iov_len;
ret = recvmsg(sock->fd, &msg, flags);
ret = recvmsg(sock->fd, &msg, flags);
- } while (ret < 0 && errno == EINTR);
+ if (ret > 0) {
+ iov[0].iov_base += ret;
+ iov[0].iov_len -= ret;
+ assert(ret <= len_last);
+ }
+ } while ((ret > 0 && ret < len_last) || (ret < 0 && errno == EINTR));
- PERROR("recvmsg inet6");
+ PERROR("recvmsg inet");
+ } else if (ret > 0) {
+ ret = len;
+ /* Else ret = 0 meaning an orderly shutdown. */
struct msghdr msg;
struct iovec iov[1];
ssize_t ret = -1;
struct msghdr msg;
struct iovec iov[1];
ssize_t ret = -1;
memset(&msg, 0, sizeof(msg));
memset(&msg, 0, sizeof(msg));
- ret = recvmsg(sock, &msg, MSG_WAITALL);
- } while (ret < 0 && errno == EINTR);
+ len_last = iov[0].iov_len;
+ ret = recvmsg(sock, &msg, 0);
+ if (ret > 0) {
+ iov[0].iov_base += ret;
+ iov[0].iov_len -= ret;
+ assert(ret <= len_last);
+ }
+ } while ((ret > 0 && ret < len_last) || (ret < 0 && errno == EINTR));
if (ret < 0) {
PERROR("recvmsg");
if (ret < 0) {
PERROR("recvmsg");
+ } else if (ret > 0) {
+ ret = len;
+ /* Else ret = 0 meaning an orderly shutdown. */