Fix: ust-comm recvmsg should handle partial receive
[lttng-ust.git] / liblttng-ust-comm / lttng-ust-comm.c
index 23de8c0949d46a1fdb140a9dc72bcc93ec5e66a0..751ad2e3cf2add1ffd805c9ca4df802034e793b1 100644 (file)
@@ -79,7 +79,6 @@ const char *lttng_ust_strerror(int code)
        if (code >= LTTNG_UST_ERR_NR)
                code = LTTNG_UST_ERR;
        return ustcomm_readable_code[USTCOMM_CODE_OFFSET(code)];
-
 }
 
 /*
@@ -117,11 +116,18 @@ int ustcomm_connect_unix_sock(const char *pathname)
        ret = connect(fd, (struct sockaddr *) &sun, sizeof(sun));
        if (ret < 0) {
                /*
-                * Don't print message on connect error, because connect
-                * is used in normal execution to detect if sessiond is
-                * alive.
+                * Don't print message on connect ENOENT error, because
+                * connect is used in normal execution to detect if
+                * sessiond is alive. ENOENT is when the unix socket
+                * file does not exist, and ECONNREFUSED is when the
+                * file exists but no sessiond is listening.
                 */
+               if (errno != ECONNREFUSED && errno != ECONNRESET
+                               && errno != ENOENT && errno != EACCES)
+                       PERROR("connect");
                ret = -errno;
+               if (ret == -ECONNREFUSED || ret == -ECONNRESET)
+                       ret = -EPIPE;
                goto error_connect;
        }
 
@@ -155,8 +161,11 @@ int ustcomm_accept_unix_sock(int sock)
        /* Blocking call */
        new_fd = accept(sock, (struct sockaddr *) &sun, &len);
        if (new_fd < 0) {
-               PERROR("accept");
-               return -errno;
+               if (errno != ECONNABORTED)
+                       PERROR("accept");
+               new_fd = -errno;
+               if (new_fd == -ECONNABORTED)
+                       new_fd = -EPIPE;
        }
        return new_fd;
 }
@@ -256,7 +265,8 @@ ssize_t ustcomm_recv_unix_sock(int sock, void *buf, size_t len)
 {
        struct msghdr msg;
        struct iovec iov[1];
-       ssize_t ret;
+       ssize_t ret = -1;
+       size_t len_last;
 
        memset(&msg, 0, sizeof(msg));
 
@@ -266,20 +276,31 @@ ssize_t ustcomm_recv_unix_sock(int sock, void *buf, size_t len)
        msg.msg_iovlen = 1;
 
        do {
+               len_last = iov[0].iov_len;
                ret = recvmsg(sock, &msg, 0);
-       } 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) {
                int shutret;
 
-               if (errno != EPIPE && errno != ECONNRESET)
+               if (errno != EPIPE && errno != ECONNRESET && errno != ECONNREFUSED)
                        PERROR("recvmsg");
                ret = -errno;
+               if (ret == -ECONNRESET || ret == -ECONNREFUSED)
+                       ret = -EPIPE;
 
                shutret = shutdown(sock, SHUT_RDWR);
                if (shutret)
                        ERR("Socket shutdown error");
+       } else if (ret > 0) {
+               ret = len;
        }
+       /* ret = 0 means an orderly shutdown. */
 
        return ret;
 }
@@ -320,6 +341,8 @@ ssize_t ustcomm_send_unix_sock(int sock, const void *buf, size_t len)
                if (errno != EPIPE && errno != ECONNRESET)
                        PERROR("sendmsg");
                ret = -errno;
+               if (ret == -ECONNRESET)
+                       ret = -EPIPE;
 
                shutret = shutdown(sock, SHUT_RDWR);
                if (shutret)
@@ -376,6 +399,9 @@ ssize_t ustcomm_send_fds_unix_sock(int sock, int *fds, size_t nb_fd)
                if (errno != EPIPE && errno != ECONNRESET) {
                        PERROR("sendmsg");
                }
+               ret = -errno;
+               if (ret == -ECONNRESET)
+                       ret = -EPIPE;
        }
        return ret;
 }
@@ -416,8 +442,9 @@ ssize_t ustcomm_recv_fds_unix_sock(int sock, int *fds, size_t nb_fd)
                if (errno != EPIPE && errno != ECONNRESET) {
                        PERROR("recvmsg fds");
                }
-               if (errno == EPIPE || errno == ECONNRESET)
-                       ret = -errno;
+               ret = -errno;
+               if (ret == -ECONNRESET)
+                       ret = -EPIPE;
                goto end;
        }
        if (ret == 0) {
@@ -502,15 +529,10 @@ int ustcomm_recv_app_reply(int sock, struct ustcomm_ust_reply *lur,
                }
                return lur->ret_code;
        default:
-               if (len < 0) {
-                       /* Transport level error */
-                       if (errno == EPIPE || errno == ECONNRESET)
-                               len = -errno;
-                       return len;
-               } else {
+               if (len >= 0) {
                        ERR("incorrect message size: %zd\n", len);
-                       return len;
                }
+               return len;
        }
 }
 
@@ -534,10 +556,12 @@ int ustcomm_send_app_cmd(int sock,
  * expected var_len.
  */
 ssize_t ustcomm_recv_channel_from_sessiond(int sock,
-               void **_chan_data, uint64_t var_len)
+               void **_chan_data, uint64_t var_len,
+               int *_wakeup_fd)
 {
        void *chan_data;
-       ssize_t len;
+       ssize_t len, nr_fd;
+       int wakeup_fd;
 
        if (var_len > LTTNG_UST_CHANNEL_DATA_MAX_LEN) {
                len = -EINVAL;
@@ -553,6 +577,18 @@ ssize_t ustcomm_recv_channel_from_sessiond(int sock,
        if (len != var_len) {
                goto error_recv;
        }
+       /* recv wakeup fd */
+       nr_fd = ustcomm_recv_fds_unix_sock(sock, &wakeup_fd, 1);
+       if (nr_fd <= 0) {
+               if (nr_fd < 0) {
+                       len = nr_fd;
+                       goto error_recv;
+               } else {
+                       len = -EIO;
+                       goto error_recv;
+               }
+       }
+       *_wakeup_fd = wakeup_fd;
        *_chan_data = chan_data;
        return len;
 
@@ -630,6 +666,26 @@ int ustcomm_send_reg_msg(int sock,
        return 0;
 }
 
+static
+int serialize_string_encoding(enum ustctl_string_encodings *ue,
+               enum lttng_string_encodings le)
+{
+       switch (le) {
+       case lttng_encode_none:
+               *ue = ustctl_encode_none;
+               break;
+       case lttng_encode_UTF8:
+               *ue = ustctl_encode_UTF8;
+               break;
+       case lttng_encode_ASCII:
+               *ue = ustctl_encode_ASCII;
+               break;
+       default:
+               return -EINVAL;
+       }
+       return 0;
+}
+
 static
 int serialize_basic_type(enum ustctl_abstract_types *uatype,
                enum lttng_abstract_types atype,
@@ -648,14 +704,17 @@ int serialize_basic_type(enum ustctl_abstract_types *uatype,
                uit->signedness = lit->signedness;
                uit->reverse_byte_order = lit->reverse_byte_order;
                uit->base = lit->base;
-               uit->encoding = lit->encoding;
+               if (serialize_string_encoding(&uit->encoding, lit->encoding))
+                       return -EINVAL;
                uit->alignment = lit->alignment;
                *uatype = ustctl_atype_integer;
                break;
        }
        case atype_string:
        {
-               ubt->string.encoding = lbt->string.encoding;
+               if (serialize_string_encoding(&ubt->string.encoding,
+                               lbt->string.encoding))
+                       return -EINVAL;
                *uatype = ustctl_atype_string;
                break;
        }
@@ -781,6 +840,47 @@ error_type:
        return ret;
 }
 
+static
+int serialize_ctx_fields(size_t *_nr_write_fields,
+               struct ustctl_field **ustctl_fields,
+               size_t nr_fields,
+               const struct lttng_ctx_field *lttng_fields)
+{
+       struct ustctl_field *fields;
+       int i, ret;
+       size_t nr_write_fields = 0;
+
+       fields = zmalloc(nr_fields * sizeof(*fields));
+       if (!fields)
+               return -ENOMEM;
+
+       for (i = 0; i < nr_fields; i++) {
+               struct ustctl_field *f;
+               const struct lttng_event_field *lf;
+
+               f = &fields[nr_write_fields];
+               lf = &lttng_fields[i].event_field;
+
+               /* skip 'nowrite' fields */
+               if (lf->nowrite)
+                       continue;
+               strncpy(f->name, lf->name, LTTNG_UST_SYM_NAME_LEN);
+               f->name[LTTNG_UST_SYM_NAME_LEN - 1] = '\0';
+               ret = serialize_one_type(&f->type, &lf->type);
+               if (ret)
+                       goto error_type;
+               nr_write_fields++;
+       }
+
+       *_nr_write_fields = nr_write_fields;
+       *ustctl_fields = fields;
+       return 0;
+
+error_type:
+       free(fields);
+       return ret;
+}
+
 /*
  * Returns 0 on success, negative error value on error.
  */
@@ -805,7 +905,7 @@ int ustcomm_register_event(int sock,
                struct ustcomm_notify_event_reply r;
        } reply;
        size_t signature_len, fields_len, model_emf_uri_len;
-       struct ustctl_field *fields;
+       struct ustctl_field *fields = NULL;
        size_t nr_write_fields = 0;
        int ret;
 
@@ -866,6 +966,8 @@ int ustcomm_register_event(int sock,
                if (len < 0) {
                        return len;
                }
+       } else {
+               free(fields);
        }
 
        if (model_emf_uri_len) {
@@ -919,7 +1021,7 @@ int ustcomm_register_channel(int sock,
        int session_objd,               /* session descriptor */
        int channel_objd,               /* channel descriptor */
        size_t nr_ctx_fields,
-       const struct lttng_event_field *ctx_fields,
+       const struct lttng_ctx_field *ctx_fields,
        uint32_t *chan_id,              /* channel id (output) */
        int *header_type)               /* header type (output) */
 {
@@ -933,7 +1035,7 @@ int ustcomm_register_channel(int sock,
                struct ustcomm_notify_channel_reply r;
        } reply;
        size_t fields_len;
-       struct ustctl_field *fields;
+       struct ustctl_field *fields = NULL;
        int ret;
        size_t nr_write_fields = 0;
 
@@ -944,7 +1046,7 @@ int ustcomm_register_channel(int sock,
 
        /* Calculate fields len, serialize fields. */
        if (nr_ctx_fields > 0) {
-               ret = serialize_fields(&nr_write_fields, &fields,
+               ret = serialize_ctx_fields(&nr_write_fields, &fields,
                                nr_ctx_fields, ctx_fields);
                if (ret)
                        return ret;
@@ -972,6 +1074,8 @@ int ustcomm_register_channel(int sock,
                if (len < 0) {
                        return len;
                }
+       } else {
+               free(fields);
        }
 
        len = ustcomm_recv_unix_sock(sock, &reply, sizeof(reply));
@@ -1015,3 +1119,43 @@ int ustcomm_register_channel(int sock,
                }
        }
 }
+
+/*
+ * Set socket reciving timeout.
+ */
+int ustcomm_setsockopt_rcv_timeout(int sock, unsigned int msec)
+{
+       int ret;
+       struct timeval tv;
+
+       tv.tv_sec = msec / 1000;
+       tv.tv_usec = (msec * 1000 % 1000000);
+
+       ret = setsockopt(sock, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+       if (ret < 0) {
+               PERROR("setsockopt SO_RCVTIMEO");
+               ret = -errno;
+       }
+
+       return ret;
+}
+
+/*
+ * Set socket sending timeout.
+ */
+int ustcomm_setsockopt_snd_timeout(int sock, unsigned int msec)
+{
+       int ret;
+       struct timeval tv;
+
+       tv.tv_sec = msec / 1000;
+       tv.tv_usec = (msec * 1000) % 1000000;
+
+       ret = setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+       if (ret < 0) {
+               PERROR("setsockopt SO_SNDTIMEO");
+               ret = -errno;
+       }
+
+       return ret;
+}
This page took 0.028119 seconds and 4 git commands to generate.