Merge pull request #16 from gierschv/feat-omap

ioctx: omap set / get / delete / clear added

rados/ioctx.go

@@ -287,3 +287,176 @@ func (ioctx *IOContext) RmXattr(oid string, name string) error {
 		return RadosError(int(ret))
 	}
 }
+
+// Append the map `pairs` to the omap `oid`
+func (ioctx *IOContext) SetOmap(oid string, pairs map[string][]byte) error {
+        c_oid := C.CString(oid)
+        defer C.free(unsafe.Pointer(c_oid))
+
+        var s C.size_t
+        var c *C.char
+        ptrSize := unsafe.Sizeof(c)
+
+        c_keys := C.malloc(C.size_t(len(pairs)) * C.size_t(ptrSize))
+        c_values := C.malloc(C.size_t(len(pairs)) * C.size_t(ptrSize))
+        c_lengths := C.malloc(C.size_t(len(pairs)) * C.size_t(unsafe.Sizeof(s)))
+
+        defer C.free(unsafe.Pointer(c_keys))
+        defer C.free(unsafe.Pointer(c_values))
+        defer C.free(unsafe.Pointer(c_lengths))
+
+        i := 0
+        for key, value := range pairs {
+            // key
+            c_key_ptr := (**C.char)(unsafe.Pointer(uintptr(c_keys) + uintptr(i) * ptrSize))
+            *c_key_ptr = C.CString(key)
+            defer C.free(unsafe.Pointer(*c_key_ptr))
+
+            // value and its length
+            c_value_ptr := (**C.char)(unsafe.Pointer(uintptr(c_values) + uintptr(i) * ptrSize))
+
+            var c_length C.size_t
+            if len(value) > 0 {
+                *c_value_ptr = (*C.char)(unsafe.Pointer(&value[0]))
+                c_length = C.size_t(len(value))
+            } else {
+                *c_value_ptr = nil
+                c_length = C.size_t(0)
+            }
+
+            c_length_ptr := (*C.size_t)(unsafe.Pointer(uintptr(c_lengths) + uintptr(i) * ptrSize))
+            *c_length_ptr = c_length
+
+            i++
+        }
+
+        op := C.rados_create_write_op()
+        C.rados_write_op_omap_set(
+                op,
+                (**C.char)(c_keys),
+                (**C.char)(c_values),
+                (*C.size_t)(c_lengths),
+                C.size_t(len(pairs)))
+
+        ret := C.rados_write_op_operate(op, ioctx.ioctx, c_oid, nil, 0)
+        C.rados_release_write_op(op)
+
+        if ret == 0 {
+                return nil
+        } else {
+                return RadosError(int(ret))
+        }
+}
+
+// Fetch a set of keys and their values from an omap and returns then as a map
+// `startAfter`: retrieve only the keys after this specified one
+// `filterPrefix`: retrieve only the keys beginning with this prefix
+// `maxReturn`: retrieve no more than `maxReturn` key/value pairs
+func (ioctx *IOContext) GetOmapValues(oid string, startAfter string, filterPrefix string, maxReturn int64) (map[string][]byte, error) {
+        c_oid := C.CString(oid)
+        c_start_after := C.CString(startAfter)
+        c_filter_prefix := C.CString(filterPrefix)
+        c_max_return := C.uint64_t(maxReturn)
+
+        defer C.free(unsafe.Pointer(c_oid))
+        defer C.free(unsafe.Pointer(c_start_after))
+        defer C.free(unsafe.Pointer(c_filter_prefix))
+
+        op := C.rados_create_read_op()
+
+        var c_iter C.rados_omap_iter_t
+        var c_prval C.int
+        C.rados_read_op_omap_get_vals(
+                op,
+                c_start_after,
+                c_filter_prefix,
+                c_max_return,
+                &c_iter,
+                &c_prval,
+        )
+
+        ret := C.rados_read_op_operate(op, ioctx.ioctx, c_oid, 0)
+        omap := map[string][]byte{}
+
+        if int(c_prval) != 0 {
+                return omap, RadosError(int(c_prval))
+        } else if int(ret) != 0 {
+                return omap, RadosError(int(ret))
+        }
+
+        for {
+                var c_key *C.char
+                var c_val *C.char
+                var c_len C.size_t
+
+                ret = C.rados_omap_get_next(c_iter, &c_key, &c_val, &c_len)
+
+                if int(ret) != 0 {
+                        return omap, RadosError(int(ret))
+                }
+
+                if c_key == nil {
+                        break
+                }
+
+                omap[C.GoString(c_key)] = C.GoBytes(unsafe.Pointer(c_val), C.int(c_len))
+        }
+
+        C.rados_omap_get_end(c_iter)
+        C.rados_release_read_op(op)
+
+        return omap, nil
+}
+
+// Remove the specified `keys` from the omap `oid`
+func (ioctx *IOContext) RmOmapKeys(oid string, keys []string) error {
+        c_oid := C.CString(oid)
+        defer C.free(unsafe.Pointer(c_oid))
+
+        var c *C.char
+        ptrSize := unsafe.Sizeof(c)
+
+        c_keys := C.malloc(C.size_t(len(keys)) * C.size_t(ptrSize))
+        defer C.free(unsafe.Pointer(c_keys))
+
+        i := 0
+        for _, key := range keys {
+                c_key_ptr := (**C.char)(unsafe.Pointer(uintptr(c_keys) + uintptr(i) * ptrSize))
+                *c_key_ptr = C.CString(key)
+                defer C.free(unsafe.Pointer(*c_key_ptr))
+                i++
+        }
+
+        op := C.rados_create_write_op()
+        C.rados_write_op_omap_rm_keys(
+                op,
+                (**C.char)(c_keys),
+                C.size_t(len(keys)))
+
+        ret := C.rados_write_op_operate(op, ioctx.ioctx, c_oid, nil, 0)
+        C.rados_release_write_op(op)
+
+        if ret == 0 {
+                return nil
+        } else {
+                return RadosError(int(ret))
+        }
+}
+
+// Clear the omap `oid`
+func (ioctx *IOContext) CleanOmap(oid string) error {
+        c_oid := C.CString(oid)
+        defer C.free(unsafe.Pointer(c_oid))
+
+        op := C.rados_create_write_op()
+        C.rados_write_op_omap_clear(op)
+
+        ret := C.rados_write_op_operate(op, ioctx.ioctx, c_oid, nil, 0)
+        C.rados_release_write_op(op)
+
+        if ret == 0 {
+                return nil
+        } else {
+                return RadosError(int(ret))
+        }
+}

rados/rados_test.go

@@ -557,3 +557,100 @@ func TestRmXattr(t *testing.T) {
 
 	pool.Destroy()
 }
+
+func TestReadWriteOmap(t *testing.T) {
+	conn, _ := rados.NewConn()
+	conn.ReadDefaultConfigFile()
+	conn.Connect()
+
+	pool_name := GetUUID()
+	err := conn.MakePool(pool_name)
+	assert.NoError(t, err)
+
+	pool, err := conn.OpenIOContext(pool_name)
+	assert.NoError(t, err)
+
+	// Set
+	orig := map[string][]byte{
+	    "key1": []byte("value1"),
+	    "key2": []byte("value2"),
+	    "prefixed-key3": []byte("value3"),
+	    "empty": []byte(""),
+	}
+
+	err = pool.SetOmap("obj", orig)
+	assert.NoError(t, err)
+
+	// Get
+	fetched, err := pool.GetOmapValues("obj", "",  "", 4)
+	assert.NoError(t, err)
+	assert.Equal(t, orig, fetched)
+
+	// Remove
+	err = pool.RmOmapKeys("obj", []string{"key1", "prefixed-key3"})
+	assert.NoError(t, err)
+
+	fetched, err = pool.GetOmapValues("obj", "",  "", 4)
+	assert.NoError(t, err)
+	assert.Equal(t, map[string][]byte{
+	    "key2": []byte("value2"),
+	    "empty": []byte(""),
+	}, fetched)
+
+	// Clear
+	err = pool.CleanOmap("obj")
+	assert.NoError(t, err)
+
+	fetched, err = pool.GetOmapValues("obj", "",  "", 4)
+	assert.NoError(t, err)
+	assert.Equal(t, map[string][]byte{}, fetched)
+
+	pool.Destroy()
+}
+
+func TestReadFilterOmap(t *testing.T) {
+	conn, _ := rados.NewConn()
+	conn.ReadDefaultConfigFile()
+	conn.Connect()
+
+	pool_name := GetUUID()
+	err := conn.MakePool(pool_name)
+	assert.NoError(t, err)
+
+	pool, err := conn.OpenIOContext(pool_name)
+	assert.NoError(t, err)
+
+	orig := map[string][]byte{
+	    "key1": []byte("value1"),
+	    "prefixed-key3": []byte("value3"),
+	    "key2": []byte("value2"),
+	}
+
+	err = pool.SetOmap("obj", orig)
+	assert.NoError(t, err)
+
+	// filter by prefix
+	fetched, err := pool.GetOmapValues("obj", "", "prefixed", 4)
+	assert.NoError(t, err)
+	assert.Equal(t, map[string][]byte{
+	    "prefixed-key3": []byte("value3"),
+	}, fetched)
+
+	// "start_after" a key
+	fetched, err = pool.GetOmapValues("obj", "key1", "", 4)
+	assert.NoError(t, err)
+	assert.Equal(t, map[string][]byte{
+	    "prefixed-key3": []byte("value3"),
+	    "key2": []byte("value2"),
+	}, fetched)
+
+	// maxReturn
+	fetched, err = pool.GetOmapValues("obj", "", "key", 1)
+	assert.NoError(t, err)
+	assert.Equal(t, map[string][]byte{
+	    "key1": []byte("value1"),
+	}, fetched)
+
+	pool.Destroy()
+}
+