libflux: use size_t for message/blob sizes

Problem: some libflux interfaces use the C int type to represent
buffer and I/O sizes, but size_t, as an unsigned type with more
range, is more appropriate in public interfaces.

Convert the following public API interfaces to use size_t:

  flux_msg_get_payload()
  flux_msg_set_payload()
  flux_request_decode_raw()
  flux_request_encode_raw()
  flux_response_decode_raw()
  flux_response_encode_raw()
  flux_respond_raw()
  flux_event_encode_raw()
  flux_event_decode_raw()
  flux_event_publish_raw()
  flux_rpc_raw()
  flux_rpc_get_raw()
  flux_kvs_lookup_get_raw()
  flux_kvs_txn_put_raw()

  Update tests and in-tree users, where appropriate.

src/broker/log.c

@@ -481,7 +481,7 @@ static void append_request_cb (flux_t *h, flux_msg_handler_t *mh,
     logbuf_t *logbuf = arg;
     uint32_t matchtag;
     const char *buf;
-    int len;
+    size_t len;
 
     if (flux_msg_get_matchtag (msg, &matchtag) < 0) {
         log_msg ("%s: malformed log request", __FUNCTION__);

src/broker/trace.c

@@ -54,7 +54,7 @@ static void trace_msg (flux_t *h,
     int type = 0;
     char buf[64];
     const char *topic = NULL;
-    int payload_size = 0;
+    size_t payload_size = 0;
     json_t *payload_json = NULL;
     int errnum = 0;
     const char *errstr = NULL;

src/cmd/builtin/content.c

@@ -23,7 +23,7 @@ static void load_to_fd (flux_t *h, int fd, const char *blobref, int flags)
 {
     flux_future_t *f;
     const uint8_t *data;
-    int size;
+    size_t size;
 
     if (!(f = content_load_byblobref (h, blobref, flags))
         || content_load_get (f, (const void **)&data, &size) < 0)

src/cmd/builtin/dump.c

@@ -154,7 +154,7 @@ static void dump_valref (struct archive *ar,
     int total_size = 0;
     struct archive_entry *entry;
     const void *data;
-    int len;
+    size_t len;
 
     /* We need the total size before we start writing archive data,
      * so make a first pass, saving the data for writing later.
@@ -220,7 +220,7 @@ static void dump_val (struct archive *ar,
 {
     struct archive_entry *entry;
     void *data;
-    int len;
+    size_t len;
 
     if (treeobj_decode_val (treeobj, &data, &len) < 0)
         log_err_exit ("%s: invalid value object", path);
@@ -296,7 +296,7 @@ static void dump_dirref (struct archive *ar,
 {
     flux_future_t *f;
     const void *buf;
-    int buflen;
+    size_t buflen;
     json_t *treeobj_deref = NULL;
 
     if (treeobj_get_count (treeobj) != 1)
@@ -356,7 +356,7 @@ static void dump_blobref (struct archive *ar,
 {
     flux_future_t *f;
     const void *buf;
-    int buflen;
+    size_t buflen;
     json_t *treeobj;
     json_t *dict;
     const char *key;

src/cmd/flux-event.c

@@ -256,7 +256,7 @@ static int event_pub (optparse_t *p, int argc, char **argv)
                 log_err_exit ("flux_recv error");
             if (optparse_hasopt (p, "raw")) {
                 const void *data;
-                int len;
+                size_t len;
                 if ((flux_event_decode_raw (msg, NULL, &data, &len) == 0
                     && match_payload_raw (payload, payloadsz, data, len)))
                     received = true;
@@ -350,7 +350,7 @@ static void event_cb (flux_t *h,
     int max_count = optparse_get_int (p, "count", 0);
     static int recv_count = 0;
     const char *payload;
-    int payloadsz;
+    size_t payloadsz;
     const char *topic;
 
     if (flux_event_decode (msg, &topic, &payload) == 0)

src/cmd/flux-kvs.c

@@ -712,7 +712,7 @@ void lookup_continuation (flux_future_t *f, void *arg)
     }
     else if (optparse_hasopt (ctx->p, "raw")) {
         const void *data;
-        int len;
+        size_t len;
         if (flux_kvs_lookup_get_raw (f, &data, &len) < 0)
             log_err_exit ("%s", key);
         if (optparse_hasopt (ctx->p, "label"))
@@ -1291,7 +1291,7 @@ static void dump_kvs_dir (const flux_kvsdir_t *dir, int maxcol,
             if (!dopt) {
                 const char *value;
                 const void *buf;
-                int len;
+                size_t len;
                 if (rootref) {
                     if (!(f = flux_kvs_lookupat (h, 0, key, rootref)))
                         log_err_exit ("%s", key);
@@ -1303,7 +1303,7 @@ static void dump_kvs_dir (const flux_kvsdir_t *dir, int maxcol,
                 if (flux_kvs_lookup_get (f, &value) == 0) // null terminated
                     dump_kvs_val (key, maxcol, value);
                 else if (flux_kvs_lookup_get_raw  (f, &buf, &len) == 0)
-                    kv_printf (key, maxcol, "%.*s", len, (char *)buf);
+                    kv_printf (key, maxcol, "%.*s", (int)len, (char *)buf);
                 else
                     log_err_exit ("%s", key);
                 flux_future_destroy (f);

src/common/libcontent/content.c

@@ -23,7 +23,7 @@
 
 flux_future_t *content_load_byhash (flux_t *h,
                                     const void *hash,
-                                    int hash_size,
+                                    size_t hash_size,
                                     int flags)
 {
     const char *topic = "content.load";
@@ -47,19 +47,19 @@ flux_future_t *content_load_byblobref (flux_t *h,
                                        int flags)
 {
     uint32_t hash[BLOBREF_MAX_DIGEST_SIZE];
-    int hash_size;
+    ssize_t hash_size;
 
     if ((hash_size = blobref_strtohash (blobref, hash, sizeof (hash))) < 0)
         return NULL;
     return content_load_byhash (h, hash, hash_size, flags);
 }
 
-int content_load_get (flux_future_t *f, const void **buf, int *len)
+int content_load_get (flux_future_t *f, const void **buf, size_t *len)
 {
     return flux_rpc_get_raw (f, buf, len);
 }
 
-flux_future_t *content_store (flux_t *h, const void *buf, int len, int flags)
+flux_future_t *content_store (flux_t *h, const void *buf, size_t len, int flags)
 {
     const char *topic = "content.store";
     uint32_t rank = FLUX_NODEID_ANY;
@@ -73,10 +73,12 @@ flux_future_t *content_store (flux_t *h, const void *buf, int len, int flags)
     return flux_rpc_raw (h, topic, buf, len, rank, 0);
 }
 
-int content_store_get_hash (flux_future_t *f, const void **hash, int *hash_size)
+int content_store_get_hash (flux_future_t *f,
+                            const void **hash,
+                            size_t *hash_size)
 {
     const void *buf;
-    int buf_size;
+    size_t buf_size;
 
     if (flux_rpc_get_raw (f, &buf, &buf_size) < 0)
         return -1;
@@ -96,7 +98,7 @@ int content_store_get_blobref (flux_future_t *f,
 
     if (!(result = flux_future_aux_get (f, auxkey))) {
         const void *hash;
-        int hash_len;
+        size_t hash_len;
         char buf[BLOBREF_MAX_STRING_SIZE];
         char *cpy = NULL;
 

src/common/libcontent/content.h

@@ -21,7 +21,7 @@ enum {
  */
 flux_future_t *content_load_byhash (flux_t *h,
                                     const void *hash,
-                                    int hash_len,
+                                    size_t hash_len,
                                     int flags);
 flux_future_t *content_load_byblobref (flux_t *h,
                                        const char *blobref,
@@ -32,17 +32,22 @@ flux_future_t *content_load_byblobref (flux_t *h,
  * Storage for 'buf' belongs to 'f' and is valid until 'f' is destroyed.
  * Returns 0 on success, -1 on failure with errno set.
  */
-int content_load_get (flux_future_t *f, const void **buf, int *len);
+int content_load_get (flux_future_t *f, const void **buf, size_t *len);
 
 /* Send request to store blob.
  */
-flux_future_t *content_store (flux_t *h, const void *buf, int len, int flags);
+flux_future_t *content_store (flux_t *h,
+                              const void *buf,
+                              size_t len,
+                              int flags);
 
 /* Get result of store request (hash or blobref).
  * Storage belongs to 'f' and is valid until 'f' is destroyed.
  * Returns 0 on success, -1 on failure with errno set.
  */
-int content_store_get_hash (flux_future_t *f, const void **hash, int *hash_len);
+int content_store_get_hash (flux_future_t *f,
+                            const void **hash,
+                            size_t *hash_len);
 int content_store_get_blobref (flux_future_t *f,
                                const char *hash_name,
                                const char **blobref);

src/common/libfilemap/filemap.c

@@ -82,7 +82,7 @@ static int extract_blob (flux_t *h,
     } entry;
     flux_future_t *f;
     const void *buf;
-    int size;
+    size_t size;
 
     if (json_unpack (o,
                      "[I,I,s]",

src/common/libflux/event.c

@@ -130,11 +130,11 @@ int flux_event_decode (const flux_msg_t *msg,
 int flux_event_decode_raw (const flux_msg_t *msg,
                            const char **topicp,
                            const void **datap,
-                           int *lenp)
+                           size_t *lenp)
 {
     const char *topic;
     const void *data = NULL;
-    int len = 0;
+    size_t len = 0;
     int rc = -1;
 
     if (!datap || !lenp) {
@@ -224,7 +224,7 @@ flux_msg_t *flux_event_encode (const char *topic, const char *s)
 
 flux_msg_t *flux_event_encode_raw (const char *topic,
                                    const void *data,
-                                   int len)
+                                   size_t len)
 {
     flux_msg_t *msg = flux_event_create (topic);
     if (!msg)
@@ -267,7 +267,7 @@ static flux_future_t *wrap_event_rpc (flux_t *h,
                                       const char *topic,
                                       int flags,
                                       const void *src,
-                                      int srclen)
+                                      size_t srclen)
 {
     flux_future_t *f;
 
@@ -371,7 +371,7 @@ flux_future_t *flux_event_publish_raw (flux_t *h,
                                        const char *topic,
                                        int flags,
                                        const void *data,
-                                       int len)
+                                       size_t len)
 {
     if (!h || !topic || (flags & ~(FLUX_MSGFLAG_PRIVATE)) != 0) {
         errno = EINVAL;

src/common/libflux/event.h

@@ -63,7 +63,7 @@ flux_msg_t *flux_event_pack (const char *topic, const char *fmt, ...);
  */
 flux_msg_t *flux_event_encode_raw (const char *topic,
                                    const void *data,
-                                   int len);
+                                   size_t len);
 
 /* Decode an event message, with optional raw payload.
  * If topic is non-NULL, assign the event topic string.
@@ -74,7 +74,7 @@ flux_msg_t *flux_event_encode_raw (const char *topic,
 int flux_event_decode_raw (const flux_msg_t *msg,
                            const char **topic,
                            const void **data,
-                           int *len);
+                           size_t *len);
 
 /* Publish an event with optional string payload.
  * The future is fulfilled once the event has been assigned a sequence number,
@@ -99,7 +99,7 @@ flux_future_t *flux_event_publish_raw (flux_t *h,
                                        const char *topic,
                                        int flags,
                                        const void *data,
-                                       int len);
+                                       size_t len);
 
 /* Obtain the event sequence number from the fulfilled
  * flux_event_publish() future.

src/common/libflux/message.c

@@ -842,7 +842,7 @@ static bool payload_overlap (flux_msg_t *msg, const void *b)
          && (char *)b <  (char *)msg->payload + msg->payload_size);
 }
 
-int flux_msg_set_payload (flux_msg_t *msg, const void *buf, int size)
+int flux_msg_set_payload (flux_msg_t *msg, const void *buf, size_t size)
 {
     if (msg_validate (msg) < 0)
         return -1;
@@ -968,7 +968,7 @@ int flux_msg_pack (flux_msg_t *msg, const char *fmt, ...)
     return rc;
 }
 
-int flux_msg_get_payload (const flux_msg_t *msg, const void **buf, int *size)
+int flux_msg_get_payload (const flux_msg_t *msg, const void **buf, size_t *size)
 {
     if (msg_validate (msg) < 0)
         return -1;
@@ -1003,7 +1003,7 @@ int flux_msg_get_string (const flux_msg_t *msg, const char **s)
 {
     const char *buf;
     const char *result;
-    int size;
+    size_t size;
 
     if (msg_validate (msg) < 0)
         return -1;
@@ -1366,7 +1366,7 @@ void flux_msg_fprint_ts (FILE *f, const flux_msg_t *msg, double timestamp)
     if (flux_msg_has_payload (msg)) {
         const char *s;
         const void *buf;
-        int size;
+        size_t size;
         if (flux_msg_get_string (msg, &s) == 0)
             fprintf (f, "%s %s\n", prefix, s);
         else if (flux_msg_get_payload (msg, &buf, &size) == 0) {

src/common/libflux/message.h

@@ -166,8 +166,10 @@ int flux_msg_get_topic (const flux_msg_t *msg, const char **topic);
  * Any old payload is deleted.
  * flux_msg_get_payload returns pointer to msg-owned buf.
  */
-int flux_msg_get_payload (const flux_msg_t *msg, const void **buf, int *size);
-int flux_msg_set_payload (flux_msg_t *msg, const void *buf, int size);
+int flux_msg_get_payload (const flux_msg_t *msg,
+                          const void **buf,
+                          size_t *size);
+int flux_msg_set_payload (flux_msg_t *msg, const void *buf, size_t size);
 bool flux_msg_has_payload (const flux_msg_t *msg);
 
 /* Test/set/clear message flags

src/common/libflux/request.c

@@ -63,11 +63,11 @@ int flux_request_decode (const flux_msg_t *msg,
 int flux_request_decode_raw (const flux_msg_t *msg,
                              const char **topic,
                              const void **data,
-                             int *len)
+                             size_t *len)
 {
     const char *ts;
     const void *d = NULL;
-    int l = 0;
+    size_t l = 0;
     int rc = -1;
 
     if (!data || !len) {
@@ -162,7 +162,7 @@ flux_msg_t *flux_request_encode (const char *topic, const char *s)
 
 flux_msg_t *flux_request_encode_raw (const char *topic,
                                      const void *data,
-                                     int len)
+                                     size_t len)
 {
     flux_msg_t *msg = request_encode (topic);
 

src/common/libflux/request.h

@@ -42,7 +42,7 @@ int flux_request_unpack (const flux_msg_t *msg,
 int flux_request_decode_raw (const flux_msg_t *msg,
                              const char **topic,
                              const void **data,
-                             int *len);
+                             size_t *len);
 
 /* Encode a request message with optional string payload.
  * If s is non-NULL, assign the string payload.
@@ -55,7 +55,7 @@ flux_msg_t *flux_request_encode (const char *topic, const char *s);
  */
 flux_msg_t *flux_request_encode_raw (const char *topic,
                                      const void *data,
-                                     int len);
+                                     size_t len);
 
 #ifdef __cplusplus
 }

src/common/libflux/response.c

@@ -71,11 +71,11 @@ int flux_response_decode (const flux_msg_t *msg,
 int flux_response_decode_raw (const flux_msg_t *msg,
                               const char **topic,
                               const void **data,
-                              int *len)
+                              size_t *len)
 {
     const char *ts;
     const void *d = NULL;
-    int l = 0;
+    size_t l = 0;
     int rc = -1;
 
     if (!data || !len) {
@@ -167,7 +167,7 @@ flux_msg_t *flux_response_encode (const char *topic, const char *s)
 
 flux_msg_t *flux_response_encode_raw (const char *topic,
                                       const void *data,
-                                      int len)
+                                      size_t len)
 {
     flux_msg_t *msg;