Merge pull request #1218 from garlick/raw_api

kvs: add flux_kvs_lookup_get_raw()

doc/man3/Makefile.am

@@ -126,13 +126,15 @@ MAN3_FILES_SECONDARY = \
 	flux_kvs_lookupat.3 \
 	flux_kvs_lookup_get.3 \
 	flux_kvs_lookup_get_unpack.3 \
+	flux_kvs_lookup_get_raw.3 \
 	flux_kvs_fence.3 \
 	flux_kvs_txn_destroy.3 \
 	flux_kvs_txn_put.3 \
 	flux_kvs_txn_pack.3 \
 	flux_kvs_txn_mkdir.3 \
 	flux_kvs_txn_unlink.3 \
-	flux_kvs_txn_symlink.3
+	flux_kvs_txn_symlink.3 \
+	flux_kvs_txn_put_raw.3
 
 ADOC_FILES  = $(MAN3_FILES_PRIMARY:%.3=%.adoc)
 XML_FILES   = $(MAN3_FILES_PRIMARY:%.3=%.xml)
@@ -231,14 +233,17 @@ flux_rpc_get.3: flux_rpc.3
 flux_rpc_get_unpack.3: flux_rpc.3
 flux_rpc_get_raw.3: flux_rpc.3
 flux_kvs_lookupat.3: flux_kvs_lookup.3
-flux_kvs_get.3: flux_kvs_lookup.3
-flux_kvs_getf.3: flux_kvs_lookup.3
+flux_kvs_lookup_get.3: flux_kvs_lookup.3
+flux_kvs_lookup_get_unpack.3: flux_kvs_lookup.3
+flux_kvs_lookup_get_raw.3: flux_kvs_lookup.3
 flux_kvs_fence.3: flux_kvs_commit.3
 flux_kvs_txn_destroy.3: flux_kvs_txn_create.3
 flux_kvs_txn_put.3: flux_kvs_txn_create.3
 flux_kvs_txn_pack.3: flux_kvs_txn_create.3
 flux_kvs_txn_mkdir.3: flux_kvs_txn_create.3
 flux_kvs_txn_unlink.3: flux_kvs_txn_create.3
+flux_kvs_txn_symlink.3: flux_kvs_txn_create.3
+flux_kvs_txn_put_raw.3: flux_kvs_txn_create.3
 
 flux_open.3: topen.c
 flux_send.3: tsend.c

doc/man3/flux_kvs_lookup.adoc

@@ -5,7 +5,7 @@ flux_kvs_lookup(3)
 
 NAME
 ----
-flux_kvs_lookup, flux_kvs_lookupat, flux_kvs_lookup_get, flux_kvs_lookup_get_unpack - look up KVS key
+flux_kvs_lookup, flux_kvs_lookupat, flux_kvs_lookup_get, flux_kvs_lookup_get_unpack, flux_kvs_lookup_get_raw - look up KVS key
 
 
 SYNOPSIS
@@ -21,6 +21,8 @@ SYNOPSIS
 
  int flux_kvs_lookup_get_unpack (flux_future_t *f, const char *fmt, ...);
 
+ int flux_kvs_lookup_get_raw (flux_future_t *f, const void **data, int *len);
+
 
 DESCRIPTION
 -----------
@@ -48,6 +50,9 @@ _json_str_ may be a JSON object, array, or bare value.
 the returned JSON is parsed according to variable arguments in Jansson
 `json_unpack()` format.
 
+`flux_kvs_lookup_get_raw()` is identical to `flux_kvs_lookup_get()` except
+the raw value is returned without decoding.
+
 These functions may be used asynchronously.
 See `flux_future_then(3)` for details.
 
@@ -75,8 +80,9 @@ RETURN VALUE
 `flux_kvs_lookup()` and `flux_kvs_lookupat()` return a
 `flux_future_t` on success, or NULL on failure with errno set appropriately.
 
-`flux_kvs_lookup_get()` and `flux_kvs_lookup_getat()`
-return 0 on success, or -1 on failure with errno set appropriately.
+`flux_kvs_lookup_get()`, `flux_kvs_lookup_get_unpack()`, and
+`flux_kvs_lookup_get_raw()` return 0 on success, or -1 on failure with
+errno set appropriately.
 
 
 ERRORS

doc/man3/flux_kvs_txn_create.adoc

@@ -5,7 +5,7 @@ flux_kvs_txn_create(3)
 
 NAME
 ----
-flux_kvs_txn_create, flux_kvs_txn_destroy, flux_kvs_txn_put, flux_kvs_txn_pack, flux_kvs_txn_mkdir, flux_kvs_txn_unlink, flux_kvs_txn_symlink - operate on a KVS transaction object
+flux_kvs_txn_create, flux_kvs_txn_destroy, flux_kvs_txn_put, flux_kvs_txn_pack, flux_kvs_txn_mkdir, flux_kvs_txn_unlink, flux_kvs_txn_symlink, flux_kvs_txn_put_raw - operate on a KVS transaction object
 
 
 SYNOPSIS
@@ -31,6 +31,10 @@ SYNOPSIS
  int flux_kvs_txn_symlink (flux_kvs_txn_t *txn, int flags,
                            const char *key, const char *target);
 
+ int flux_kvs_txn_put_raw (flux_kvs_txn_t *txn, int flags,
+                           const char *key, const void *data, int len);
+
+
 
 DESCRIPTION
 -----------
@@ -65,6 +69,9 @@ all its contents are removed as well.
 `flux_kvs_txn_symlink()` sets _key_ to a symbolic link pointing to _target_,
 another key.  _target_ need not exist.
 
+`flux_kvs_txn_put_raw()` sets _key_ to a value containing raw data
+referred to by _data_ of length _len_.
+
 
 FLAGS
 -----
@@ -90,8 +97,8 @@ RETURN VALUE
 or NULL on failure with errno set appropriately.
 
 `flux_kvs_txn_put()`, `flux_kvs_txn_pack()`, `flux_kvs_txn_mkdir()`,
-`flux_kvs_txn_unlink()`, and `flux_kvs_txn_symlink()` returns 0 on success,
-or -1 on failure with errno set appropriately.
+`flux_kvs_txn_unlink()`, `flux_kvs_txn_symlink()`, and `flux_kvs_txn_put_raw()`
+returns 0 on success, or -1 on failure with errno set appropriately.
 
 ERRORS
 ------

src/common/libkvs/kvs_lookup.c

@@ -244,6 +244,32 @@ int flux_kvs_lookup_get_unpack (flux_future_t *f, const char *fmt, ...)
     return rc;
 }
 
+int flux_kvs_lookup_get_raw (flux_future_t *f, const void **data, int *len)
+{
+    struct lookup_ctx *ctx;
+
+    if (!(ctx = flux_future_aux_get (f, auxkey))) {
+        errno = EINVAL;
+        return -1;
+    }
+    if (!(ctx->treeobj)) {
+        if (flux_rpc_get_unpack (f, "{s:o}", "val", &ctx->treeobj) < 0) {
+            errno = EINVAL;
+            return -1;
+        }
+    }
+    if (!ctx->val_valid) {
+        if (treeobj_decode_val (ctx->treeobj, &ctx->val_data,
+                                              &ctx->val_len) < 0)
+            return -1;
+        ctx->val_valid = true;
+    }
+    if (data)
+        *data = ctx->val_data;
+    if (len)
+        *len = ctx->val_len;
+    return 0;
+}
 
 /*
  * vi:tabstop=4 shiftwidth=4 expandtab

src/common/libkvs/kvs_lookup.h

@@ -13,6 +13,7 @@ flux_future_t *flux_kvs_lookupat (flux_t *h, int flags, const char *key,
 
 int flux_kvs_lookup_get (flux_future_t *f, const char **json_str);
 int flux_kvs_lookup_get_unpack (flux_future_t *f, const char *fmt, ...);
+int flux_kvs_lookup_get_raw (flux_future_t *f, const void **data, int *len);
 
 #endif /* !_FLUX_CORE_KVS_LOOKUP_H */
 

src/common/libkvs/kvs_txn.c

@@ -148,7 +148,7 @@ static int flux_kvs_txn_put_treeobj (flux_kvs_txn_t *txn, int flags,
 }
 
 int flux_kvs_txn_put_raw (flux_kvs_txn_t *txn, int flags,
-                          const char *key, void *data, int len)
+                          const char *key, const void *data, int len)
 {
     json_t *dirent = NULL;
     int saved_errno;

src/common/libkvs/kvs_txn.h

@@ -13,7 +13,7 @@ int flux_kvs_txn_pack (flux_kvs_txn_t *txn, int flags,
                        const char *key, const char *fmt, ...);
 
 int flux_kvs_txn_put_raw (flux_kvs_txn_t *txn, int flags,
-                          const char *key, void *data, int len);
+                          const char *key, const void *data, int len);
 
 int flux_kvs_txn_mkdir (flux_kvs_txn_t *txn, int flags,
                         const char *key);

src/common/libkvs/test/kvs_lookup.c

@@ -28,6 +28,10 @@ void errors (void)
     errno = 0;
     ok (flux_kvs_lookup_get_unpack (NULL, NULL) < 0 && errno == EINVAL,
         "flux_kvs_lookup_get_unpack fails on bad input");
+
+    errno = 0;
+    ok (flux_kvs_lookup_get_raw (NULL, NULL, NULL) < 0 && errno == EINVAL,
+        "flux_kvs_lookup_get_raw fails on bad input");
 }
 
 int main (int argc, char *argv[])

t/kvs/basic.c

@@ -34,7 +34,6 @@
 
 #include "src/common/libutil/xzmalloc.h"
 #include "src/common/libutil/log.h"
-#include "src/common/libutil/base64.h"
 #include "src/common/libutil/readall.h"
 
 
@@ -274,27 +273,21 @@ void cmd_copy_tokvs (flux_t *h, int argc, char **argv)
             log_err_exit ("%s", file);
         (void)close (fd);
     }
-    int s_len = base64_encode_length (len);
-    char *s_buf = xzmalloc (s_len);
-    if (base64_encode_block (s_buf, &s_len, buf, len) < 0)
-        log_err_exit ("base64_encode_block error");
     if (!(txn = flux_kvs_txn_create ()))
         log_err_exit ("flux_kvs_txn_create");
-    if (flux_kvs_txn_pack (txn, 0, key, "{s:s}", "data", s_buf) < 0)
-        log_err_exit ("flux_kvs_txn_pack");
+    if (flux_kvs_txn_put_raw (txn, 0, key, buf, len) < 0)
+        log_err_exit ("flux_kvs_txn_put_raw");
     if (!(f = flux_kvs_commit (h, 0, txn)) || flux_future_get (f, NULL) < 0)
         log_err_exit ("flux_kvs_commit");
     flux_kvs_txn_destroy (txn);
     free (buf);
-    free (s_buf);
 }
 
 void cmd_copy_fromkvs (flux_t *h, int argc, char **argv)
 {
     char *file, *key;
-    int fd, len, s_len;
-    uint8_t *buf;
-    const char *s_buf;
+    int fd, len;
+    const uint8_t *buf;
     flux_future_t *f;
 
     if (argc != 2)
@@ -303,13 +296,8 @@ void cmd_copy_fromkvs (flux_t *h, int argc, char **argv)
     file = argv[1];
     if (!(f = flux_kvs_lookup (h, 0, key)))
         log_err_exit ("flux_kvs_lookup");
-    if (flux_kvs_lookup_get_unpack (f, "{s:s}", "data", &s_buf) < 0)
+    if (flux_kvs_lookup_get_raw (f, (const void **)&buf, &len) < 0)
         log_err_exit ("%s", key);
-    s_len = strlen (s_buf);
-    len = base64_decode_length (s_len);
-    buf = xzmalloc (len);
-    if (base64_decode_block (buf, &len, s_buf, s_len) < 0)
-        log_err_exit ("%s: base64 decode error", key);
     if (!strcmp (file, "-")) {
         if (write_all (STDOUT_FILENO, buf, len) < 0)
             log_err_exit ("stdout");
@@ -321,7 +309,6 @@ void cmd_copy_fromkvs (flux_t *h, int argc, char **argv)
         if (close (fd) < 0)
             log_err_exit ("%s", file);
     }
-    free (buf);
     flux_future_destroy (f);
 }
 

t/t1002-kvs-extra.t

@@ -261,6 +261,12 @@ test_expect_success 'kvs: getat: works on outdated root' '
 	test $(${KVSBASIC} getat $ROOTREF $TEST.a.b.c) = 42
 '
 
+test_expect_success 'kvs: zero size raw value can be stored and retrieved' '
+	flux kvs unlink -Rf $TEST &&
+	${KVSBASIC} copy-tokvs $TEST.empty -  </dev/null &&
+	test $(${KVSBASIC} copy-fromkvs $TEST.empty -|wc -c) -eq 0
+'
+
 test_expect_success 'kvs: kvsdir_get_size works' '
 	flux kvs mkdir $TEST.dirsize &&
 	flux kvs put $TEST.dirsize.a=1 &&
@@ -284,10 +290,9 @@ test_expect_success 'kvs: large put stores raw data into content store' '
         flux content load ${largevalhash} | grep $largeval
 '
 
-# TODO - convert to using "flux content store", see issue1216
 test_expect_success 'kvs: valref that points to content store data can be read' '
         flux kvs unlink -Rf $TEST &&
- 	flux kvs put $TEST.largeval=$largeval &&
+	echo "$largeval" | flux content store &&
 	${KVSBASIC} put-treeobj $TEST.largeval2="{\"data\":[\"${largevalhash}\"],\"type\":\"valref\",\"ver\":1}" &&
         flux kvs get $TEST.largeval2 | grep $largeval
 '