Repository: ceph/go-ceph
Pull request #408 (2020-12-02 16:56:05)

rados: add a framework for read and write operations (new attempt)

This is a redo of what was found in #315. Please see below for a more detailed list of what is different from that PR.


The Ceph librados API provides a suite of API calls for interacting with "operations". These are sets of steps that put together perform a batch operation atomically. The API flow goes a bit like (in pseudocode):

```
op = create_op()
assert_version(op, ...)
read_omap(op, ...)
read_data(op, ...)
operate(op)
```

In go-ceph we were already using some of these API calls, but we did not expose the batch nature of them. Rather, we would wrap it all in one Go function and effecitvely hid this. While convenient if all you want to do is read or write the omap or "touch" an object, it prevents the user of go-ceph from using the "full power" of the APIs ceph provides.

My goal with this PR is to establish a basic pattern of use in our Go APIs that reflects the power of the ceph APIs in a way that is fairly idiomatic to Go - much like our general approach elsewhere in the library.

In addition, since these are multiple calls that could be spread out over time the new code must properly manage any C memory that the librados APIs need. Some calls expect a pointer to a an array of strings and it could be hours between the time this memory is allocated and it is used by the operate call. This means part of the design in these patches is to support the use of the C memory but ease the management.

For both read and write operations the code flows as such (more Go-ish pseudo code):

```go
op := CreateWriteOp()
defer op.Release()
op.Create()
op.Write(...)
op.SetOmap(...)
err := op.Operate(ioctx, "myoid", OperationNoFlags)
```

In some cases the steps are expected to acquire data when the operation is performed. The functions that add those steps to the operation will return. The returned "step objects" can be used to get data after the operation:

```go
op := CreateWriteOp()
defer op.Release()
op.AssertExists()
omapGetter1 := op.GetOmapValues(...)
err := op.Operate(ioctx, "myoid", OperationNoFlags)
// simplified! check for errors!
for {
    kv, _ := ommapGetter1.Next()
    if kv == nil {
       break
    }
    // do something with the key-value pair
}
```

In general, each step may return a type with a different set of functions but the idea is that the operation is performed and then the return from the step can be used to access data.


This PR does not contain a complete set of functions for either read ops or write ops. It is just enough (IMO) to demonstrate how future operation steps can be added and convert most of the existing uses of operation calls to
this new approach.


## Major differences from PR #315

* What was called an element is now called a step.
* Tweaked and simplified error type.
* Removed the concept of "reset". It turns out that this idea was based on
  a faulty understanding of the Ceph calls. Because ceph level operations can
  not be "used" more than once for a given object, it made no sense to
  try and support it in go-ceph. This futher simplified the code.
* The patch ordering is similar but I have reorganized it a bit to hopefully
  ease reviewing.
* A few new (very simple) test cases have been added.

## Work (intentionally) left undone

* There are some TODOs left in doc comments for ceph flags. It's probably worth
  calling out the most important of these but this will require reading ceph sources.
* Write operations are more fleshed out than read operation calls. This is due
  to reads being a bit more complex than writes to implement and the desire
  not to boil the ocean with this PR.
* The OperationError type is left fairly simplistic and no "helper" methods
  remain in the code. I expect that there will be a use case for something
  like Unwrap for this type but it can be added later.




## Checklist
- [x] Added tests for features and functional changes
- [x] Public functions and types are documented
- [x] Standard formatting is applied to Go code


phlogistonjohn: Dear team,
I realize that this is fairly close to release but this is essentially a reworked version of #315. It is my sincere wish that we do our best to try and land this in v0.7. I do realize this is a largish series - but if you have helped review #315 most of it should look familiar. I tried to incorporate as much of the feedback on that PR as possible.

Thanks!

(CC: @ansiwen @agarwal-mudit @nixpanic )
ansiwen: I'm still not done yet, but I found at least one issue.
phlogistonjohn: I have pushed a new version of the patches taking into account @ansiwen review so far. Thanks!
phlogistonjohn: I have updated the PR to take the previously outstanding feedback. I closed as many of the subtreads as I could so that it was clear that feedback had been acted on or considered. If a topic needs to be revisited, feel free, I just didn't want lots of stale sub-threads hanging open.

PTAL @ansiwen @nixpanic 

Patch:
diff --git a/internal/timespec/timespec.go b/internal/timespec/timespec.go
index 54076e3..1e4b09a 100644
--- a/internal/timespec/timespec.go
+++ b/internal/timespec/timespec.go
@@ -28,3 +28,12 @@ func CStructToTimespec(cts CTimespecPtr) Timespec {
 		Nsec: int64(t.tv_nsec),
 	}
 }
+
+// CopyToCStruct copies the time values from a Timespec to a previously
+// allocated C `struct timespec`. Due to restrictions on Cgo the C pointer
+// must be passed via the CTimespecPtr wrapper.
+func CopyToCStruct(ts Timespec, cts CTimespecPtr) {
+	t := (*C.struct_timespec)(cts)
+	t.tv_sec = C.time_t(ts.Sec)
+	t.tv_nsec = C.long(ts.Nsec)
+}
diff --git a/rados/errors.go b/rados/errors.go
index 6039d12..d0de0c8 100644
--- a/rados/errors.go
+++ b/rados/errors.go
@@ -52,6 +52,9 @@ var (
 	// ErrInvalidIOContext may be returned if an api call requires an IOContext
 	// but IOContext is not ready for use.
 	ErrInvalidIOContext = errors.New("IOContext is not ready for use")
+	// ErrOperationIncomplete is returned from write op or read op steps for
+	// which the operation has not been performed yet.
+	ErrOperationIncomplete = errors.New("Operation has not been performed yet")
 )
 
 // Public radosErrors:
diff --git a/rados/ioctx.go b/rados/ioctx.go
index d7a5d21..b3462e6 100644
--- a/rados/ioctx.go
+++ b/rados/ioctx.go
@@ -128,15 +128,10 @@ func (ioctx *IOContext) SetNamespace(namespace string) {
 //  void rados_write_op_create(rados_write_op_t write_op, int exclusive,
 //                             const char* category)
 func (ioctx *IOContext) Create(oid string, exclusive CreateOption) error {
-	c_oid := C.CString(oid)
-	defer C.free(unsafe.Pointer(c_oid))
-
-	op := C.rados_create_write_op()
-	C.rados_write_op_create(op, C.int(exclusive), nil)
-	ret := C.rados_write_op_operate(op, ioctx.ioctx, c_oid, nil, 0)
-	C.rados_release_write_op(op)
-
-	return getError(ret)
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(exclusive)
+	return op.operateCompat(ioctx, oid)
 }
 
 // Write writes len(data) bytes to the object with key oid starting at byte
diff --git a/rados/omap.go b/rados/omap.go
index 7d279c2..79b4c8e 100644
--- a/rados/omap.go
+++ b/rados/omap.go
@@ -1,69 +1,230 @@
 package rados
 
-// #cgo LDFLAGS: -lrados
-// #include <stdlib.h>
-// #include <rados/librados.h>
-//
+/*
+#cgo LDFLAGS: -lrados
+#include <stdlib.h>
+#include <rados/librados.h>
+
+typedef void* voidptr;
+
+*/
 import "C"
 
 import (
+	"runtime"
 	"unsafe"
 )
 
-// SetOmap appends 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))
+const (
+	ptrSize   = C.sizeof_voidptr
+	sizeTSize = C.sizeof_size_t
+)
 
-	var s C.size_t
-	var c *C.char
-	ptrSize := unsafe.Sizeof(c)
+// setOmapStep is a write op step. It holds C memory used in the operation.
+type setOmapStep struct {
+	withRefs
+	withoutUpdate
 
-	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)))
+	// C arguments
+	cKeys    **C.char
+	cValues  **C.char
+	cLengths *C.size_t
+	cNum     C.size_t
+}
 
-	defer C.free(unsafe.Pointer(c_keys))
-	defer C.free(unsafe.Pointer(c_values))
-	defer C.free(unsafe.Pointer(c_lengths))
+func newSetOmapStep(pairs map[string][]byte) *setOmapStep {
 
-	i := 0
+	maplen := C.size_t(len(pairs))
+	cKeys := C.malloc(maplen * ptrSize)
+	cValues := C.malloc(maplen * ptrSize)
+	cLengths := C.malloc(maplen * sizeTSize)
+
+	sos := &setOmapStep{
+		cKeys:    (**C.char)(cKeys),
+		cValues:  (**C.char)(cValues),
+		cLengths: (*C.size_t)(cLengths),
+		cNum:     C.size_t(len(pairs)),
+	}
+	sos.add(cKeys)
+	sos.add(cValues)
+	sos.add(cLengths)
+
+	var i uintptr
 	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))
+		ck := C.CString(key)
+		sos.add(unsafe.Pointer(ck))
+		ckp := (**C.char)(unsafe.Pointer(uintptr(cKeys) + i*ptrSize))
+		*ckp = ck
 
 		// 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))
+		cvp := (**C.char)(unsafe.Pointer(uintptr(cValues) + i*ptrSize))
+		vlen := C.size_t(len(value))
+		if vlen > 0 {
+			cv := C.CBytes(value)
+			sos.add(cv)
+			*cvp = (*C.char)(cv)
 		} else {
-			*c_value_ptr = nil
-			c_length = C.size_t(0)
+			*cvp = nil
 		}
 
-		c_length_ptr := (*C.size_t)(unsafe.Pointer(uintptr(c_lengths) + uintptr(i)*ptrSize))
-		*c_length_ptr = c_length
+		clp := (*C.size_t)(unsafe.Pointer(uintptr(cLengths) + i*ptrSize))
+		*clp = vlen
 
 		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)))
+	runtime.SetFinalizer(sos, opStepFinalizer)
+	return sos
+}
 
-	ret := C.rados_write_op_operate(op, ioctx.ioctx, c_oid, nil, 0)
-	C.rados_release_write_op(op)
+func (sos *setOmapStep) free() {
+	sos.cKeys = nil
+	sos.cValues = nil
+	sos.cLengths = nil
+	sos.withRefs.free()
+}
 
-	return getError(ret)
+// OmapKeyValue items are returned by the GetOmapStep's Next call.
+type OmapKeyValue struct {
+	Key   string
+	Value []byte
+}
+
+// GetOmapStep values are used to get the results of an GetOmapValues call
+// on a WriteOp. Until the Operate method of the WriteOp is called the Next
+// call will return an error. After Operate is called, the Next call will
+// return valid results.
+//
+// The life cycle of the GetOmapStep is bound to the ReadOp, if the ReadOp
+// Release method is called the public methods of the step must no longer be
+// used and may return errors.
+type GetOmapStep struct {
+	// inputs:
+	startAfter   string
+	filterPrefix string
+	maxReturn    uint64
+
+	// arguments:
+	cStartAfter   *C.char
+	cFilterPrefix *C.char
+
+	// C returned data:
+	iter C.rados_omap_iter_t
+	more C.uchar
+	rval C.int
+
+	// internal state:
+
+	// canIterate is only set after the operation is performed and is
+	// intended to prevent premature fetching of data
+	canIterate bool
+}
+
+func newGetOmapStep(startAfter, filterPrefix string, maxReturn uint64) *GetOmapStep {
+	gos := &GetOmapStep{
+		startAfter:    startAfter,
+		filterPrefix:  filterPrefix,
+		maxReturn:     maxReturn,
+		cStartAfter:   C.CString(startAfter),
+		cFilterPrefix: C.CString(filterPrefix),
+	}
+	runtime.SetFinalizer(gos, opStepFinalizer)
+	return gos
+}
+
+func (gos *GetOmapStep) free() {
+	gos.canIterate = false
+	if gos.iter != nil {
+		C.rados_omap_get_end(gos.iter)
+	}
+	gos.iter = nil
+	gos.more = 0
+	gos.rval = 0
+	C.free(unsafe.Pointer(gos.cStartAfter))
+	gos.cStartAfter = nil
+	C.free(unsafe.Pointer(gos.cFilterPrefix))
+	gos.cFilterPrefix = nil
+}
+
+func (gos *GetOmapStep) update() error {
+	err := getError(gos.rval)
+	gos.canIterate = (err == nil)
+	return err
+}
+
+// Next returns the next key value pair or nil if iteration is exhausted.
+func (gos *GetOmapStep) Next() (*OmapKeyValue, error) {
+	if !gos.canIterate {
+		return nil, ErrOperationIncomplete
+	}
+	var (
+		cKey *C.char
+		cVal *C.char
+		cLen C.size_t
+	)
+	ret := C.rados_omap_get_next(gos.iter, &cKey, &cVal, &cLen)
+	if ret != 0 {
+		return nil, getError(ret)
+	}
+	if cKey == nil {
+		return nil, nil
+	}
+	return &OmapKeyValue{
+		Key:   C.GoString(cKey),
+		Value: C.GoBytes(unsafe.Pointer(cVal), C.int(cLen)),
+	}, nil
+}
+
+// More returns true if there are more matching keys available.
+func (gos *GetOmapStep) More() bool {
+	// tad bit hacky, but go can't automatically convert from
+	// unsigned char to bool
+	return gos.more != 0
+}
+
+// removeOmapKeysStep is a write operation step used to track state, especially
+// C memory, across the setup and use of a WriteOp.
+type removeOmapKeysStep struct {
+	withRefs
+	withoutUpdate
+
+	// arguments:
+	cKeys **C.char
+	cNum  C.size_t
+}
+
+func newRemoveOmapKeysStep(keys []string) *removeOmapKeysStep {
+	cKeys := C.malloc(C.size_t(len(keys)) * ptrSize)
+	roks := &removeOmapKeysStep{
+		cKeys: (**C.char)(cKeys),
+		cNum:  C.size_t(len(keys)),
+	}
+	roks.add(cKeys)
+
+	i := 0
+	for _, key := range keys {
+		ckp := (**C.char)(unsafe.Pointer(uintptr(cKeys) + uintptr(i)*ptrSize))
+		*ckp = C.CString(key)
+		roks.add(unsafe.Pointer(*ckp))
+		i++
+	}
+
+	runtime.SetFinalizer(roks, opStepFinalizer)
+	return roks
+}
+
+func (roks *removeOmapKeysStep) free() {
+	roks.cKeys = nil
+	roks.withRefs.free()
+}
+
+// SetOmap appends the map `pairs` to the omap `oid`
+func (ioctx *IOContext) SetOmap(oid string, pairs map[string][]byte) error {
+	op := CreateWriteOp()
+	defer op.Release()
+	op.SetOmap(pairs)
+	return op.operateCompat(ioctx, oid)
 }
 
 // OmapListFunc is the type of the function called for each omap key
@@ -78,58 +239,25 @@ type OmapListFunc func(key string, value []byte)
 // `maxReturn`: iterate no more than `maxReturn` key/value pairs
 // `listFn`: the function called at each iteration
 func (ioctx *IOContext) ListOmapValues(oid string, startAfter string, filterPrefix string, maxReturn int64, listFn OmapListFunc) 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_vals2(
-		op,
-		c_start_after,
-		c_filter_prefix,
-		c_max_return,
-		&c_iter,
-		nil,
-		&c_prval,
-	)
-
-	ret := C.rados_read_op_operate(op, ioctx.ioctx, c_oid, 0)
 
-	if int(ret) != 0 {
-		return getError(ret)
-	} else if int(c_prval) != 0 {
-		return getError(c_prval)
+	op := CreateReadOp()
+	defer op.Release()
+	gos := op.GetOmapValues(startAfter, filterPrefix, uint64(maxReturn))
+	err := op.operateCompat(ioctx, oid)
+	if err != nil {
+		return err
 	}
 
 	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 getError(ret)
+		kv, err := gos.Next()
+		if err != nil {
+			return err
 		}
-
-		if c_key == nil {
+		if kv == nil {
 			break
 		}
-
-		listFn(C.GoString(c_key), C.GoBytes(unsafe.Pointer(c_val), C.int(c_len)))
+		listFn(kv.Key, kv.Value)
 	}
-
-	C.rados_omap_get_end(c_iter)
-	C.rados_release_read_op(op)
-
 	return nil
 }
 
@@ -184,45 +312,16 @@ func (ioctx *IOContext) GetAllOmapValues(oid string, startAfter string, filterPr
 
 // RmOmapKeys removes 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)
-
-	return getError(ret)
+	op := CreateWriteOp()
+	defer op.Release()
+	op.RmOmapKeys(keys)
+	return op.operateCompat(ioctx, oid)
 }
 
 // CleanOmap clears 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)
-
-	return getError(ret)
+	op := CreateWriteOp()
+	defer op.Release()
+	op.CleanOmap()
+	return op.operateCompat(ioctx, oid)
 }
diff --git a/rados/operation.go b/rados/operation.go
new file mode 100644
index 0000000..ca3ab38
--- /dev/null
+++ b/rados/operation.go
@@ -0,0 +1,151 @@
+package rados
+
+// #include <stdlib.h>
+import "C"
+
+import (
+	"fmt"
+	"strings"
+	"unsafe"
+)
+
+// The file operation.go exists to support both read op and write op types that
+// have some pretty common behaviors between them. In C/C++ its assumed that
+// the buffer types and other pointers will not be freed between passing them
+// to the action setup calls (things like rados_write_op_write or
+// rados_read_op_omap_get_vals2) and the call to Operate(...).  Since there's
+// nothing stopping one from sleeping for hours between these calls, or passing
+// the op to other functions and calling Operate there, we want a mechanism
+// that's (fairly) simple to understand and won't run afoul of Go's garbage
+// collection.  That's one reason the operation type tracks the steps (the
+// parts that track complex inputs and outputs) so that as long as the op
+// exists it will have a reference to the step, which will have references
+// to the C language types.
+
+type opKind string
+
+const (
+	readOp  opKind = "read"
+	writeOp opKind = "write"
+)
+
+// OperationError is an error type that may be returned by an Operate call.
+// It captures the error from the operate call itself and any errors from
+// steps that can return an error.
+type OperationError struct {
+	kind       opKind
+	OpError    error
+	StepErrors map[int]error
+}
+
+func (e OperationError) Error() string {
+	subErrors := []string{}
+	if e.OpError != nil {
+		subErrors = append(subErrors,
+			fmt.Sprintf("op=%s", e.OpError))
+	}
+	for idx, es := range e.StepErrors {
+		subErrors = append(subErrors,
+			fmt.Sprintf("Step#%d=%s", idx, es))
+	}
+	return fmt.Sprintf(
+		"%s operation error: %s",
+		e.kind,
+		strings.Join(subErrors, ", "))
+}
+
+// opStep provides an interface for types that are tied to the management of
+// data being input or output from write ops and read ops. The steps are
+// meant to simplify the internals of the ops themselves and be exportable when
+// appropriate. If a step is not being exported it should not be returned
+// from an ops action function. If the step is exported it should be
+// returned from an ops action function.
+//
+// Not all types implementing opStep are expected to need all the functions
+// in the interface. However, for the sake of simplicity on the op side, we use
+// the same interface for all cases and expect those implementing opStep
+// just embed the without* types that provide no-op implementation of
+// functions that make up this interface.
+type opStep interface {
+	// update the state of the step after the call to Operate.
+	// It can be used to convert values from C and cache them and/or
+	// communicate a failure of the action associated with the step.  The
+	// update call will only be made once. Implementations are not required to
+	// handle this call being made more than once.
+	update() error
+	// free will be called to free any resources, especially C memory, that
+	// the step is managing. The behavior of free should be idempotent and
+	// handle being called more than once.
+	free()
+}
+
+// operation represents some of the shared underlying mechanisms for
+// both read and write op types.
+type operation struct {
+	steps []opStep
+}
+
+// free will call the free method of all the steps this operation
+// contains.
+func (o *operation) free() {
+	for i := range o.steps {
+		o.steps[i].free()
+	}
+}
+
+// update the operation and the steps it contains. The top-level result
+// of the rados call is passed in as ret and used to construct errors.
+// The update call of each step is used to update the contents of each
+// step and gather any errors from those steps.
+func (o *operation) update(kind opKind, ret C.int) error {
+	stepErrors := map[int]error{}
+	for i := range o.steps {
+		if err := o.steps[i].update(); err != nil {
+			stepErrors[i] = err
+		}
+	}
+	if ret == 0 && len(stepErrors) == 0 {
+		return nil
+	}
+	return OperationError{
+		kind:       kind,
+		OpError:    getError(ret),
+		StepErrors: stepErrors,
+	}
+}
+
+func opStepFinalizer(s opStep) {
+	if s != nil {
+		s.free()
+	}
+}
+
+// withoutUpdate can be embedded in a struct to help indicate
+// the type implements the opStep interface but has a no-op
+// update function.
+type withoutUpdate struct{}
+
+func (*withoutUpdate) update() error { return nil }
+
+// withoutFree can be embedded in a struct to help indicate
+// the type implements the opStep interface but has a no-op
+// free function.
+type withoutFree struct{}
+
+func (*withoutFree) free() {}
+
+// withRefs is a embeddable type to help track and free C memory.
+type withRefs struct {
+	refs []unsafe.Pointer
+}
+
+func (w *withRefs) free() {
+	for i := range w.refs {
+		C.free(w.refs[i])
+	}
+	w.refs = nil
+}
+
+func (w *withRefs) add(ptr unsafe.Pointer) {
+	w.refs = append(w.refs, ptr)
+}
diff --git a/rados/operation_flags.go b/rados/operation_flags.go
new file mode 100644
index 0000000..957f4f2
--- /dev/null
+++ b/rados/operation_flags.go
@@ -0,0 +1,37 @@
+package rados
+
+// #cgo LDFLAGS: -lrados
+// #include <errno.h>
+// #include <stdlib.h>
+// #include <rados/librados.h>
+//
+import "C"
+
+// OperationFlags control the behavior of read and write operations.
+type OperationFlags int
+
+const (
+	// OperationNoFlag indicates no special behavior is requested.
+	OperationNoFlag = OperationFlags(C.LIBRADOS_OPERATION_NOFLAG)
+	// OperationBalanceReads TODO
+	OperationBalanceReads = OperationFlags(C.LIBRADOS_OPERATION_BALANCE_READS)
+	// OperationLocalizeReads TODO
+	OperationLocalizeReads = OperationFlags(C.LIBRADOS_OPERATION_LOCALIZE_READS)
+	// OperationOrderReadsWrites TODO
+	OperationOrderReadsWrites = OperationFlags(C.LIBRADOS_OPERATION_ORDER_READS_WRITES)
+	// OperationIgnoreCache TODO
+	OperationIgnoreCache = OperationFlags(C.LIBRADOS_OPERATION_IGNORE_CACHE)
+	// OperationSkipRWLocks TODO
+	OperationSkipRWLocks = OperationFlags(C.LIBRADOS_OPERATION_SKIPRWLOCKS)
+	// OperationIgnoreOverlay TODO
+	OperationIgnoreOverlay = OperationFlags(C.LIBRADOS_OPERATION_IGNORE_OVERLAY)
+	// OperationFullTry send request to a full cluster or pool, ops such as delete
+	// can succeed while other ops will return out-of-space errors.
+	OperationFullTry = OperationFlags(C.LIBRADOS_OPERATION_FULL_TRY)
+	// OperationFullForce TODO
+	OperationFullForce = OperationFlags(C.LIBRADOS_OPERATION_FULL_FORCE)
+	// OperationIgnoreRedirect TODO
+	OperationIgnoreRedirect = OperationFlags(C.LIBRADOS_OPERATION_IGNORE_REDIRECT)
+	// OperationOrderSnap TODO
+	OperationOrderSnap = OperationFlags(C.LIBRADOS_OPERATION_ORDERSNAP)
+)
diff --git a/rados/read_op.go b/rados/read_op.go
new file mode 100644
index 0000000..4f95e63
--- /dev/null
+++ b/rados/read_op.go
@@ -0,0 +1,84 @@
+package rados
+
+// #cgo LDFLAGS: -lrados
+// #include <errno.h>
+// #include <stdlib.h>
+// #include <rados/librados.h>
+//
+import "C"
+
+import (
+	"unsafe"
+)
+
+// ReadOp manages a set of discrete object read actions that will be performed
+// together atomically.
+type ReadOp struct {
+	operation
+	op C.rados_read_op_t
+}
+
+// CreateReadOp returns a newly constructed read operation.
+func CreateReadOp() *ReadOp {
+	return &ReadOp{
+		op: C.rados_create_read_op(),
+	}
+}
+
+// Release the resources associated with this read operation.
+func (r *ReadOp) Release() {
+	C.rados_release_read_op(r.op)
+	r.op = nil
+	r.free()
+}
+
+// Operate will perform the operation(s).
+func (r *ReadOp) Operate(ioctx *IOContext, oid string, flags OperationFlags) error {
+	if err := ioctx.validate(); err != nil {
+		return err
+	}
+
+	cOid := C.CString(oid)
+	defer C.free(unsafe.Pointer(cOid))
+
+	ret := C.rados_read_op_operate(r.op, ioctx.ioctx, cOid, C.int(flags))
+	return r.update(readOp, ret)
+}
+
+func (r *ReadOp) operateCompat(ioctx *IOContext, oid string) error {
+	switch err := r.Operate(ioctx, oid, OperationNoFlag).(type) {
+	case nil:
+		return nil
+	case OperationError:
+		return err.OpError
+	default:
+		return err
+	}
+}
+
+// AssertExists assures the object targeted by the read op exists.
+//
+// Implements:
+//  void rados_read_op_assert_exists(rados_read_op_t read_op);
+func (r *ReadOp) AssertExists() {
+	C.rados_read_op_assert_exists(r.op)
+}
+
+// GetOmapValues is used to iterate over a set, or sub-set, of omap keys
+// as part of a read operation. An GetOmapStep is returned from this
+// function. The GetOmapStep may be used to iterate over the key-value
+// pairs after the Operate call has been performed.
+func (r *ReadOp) GetOmapValues(startAfter, filterPrefix string, maxReturn uint64) *GetOmapStep {
+	gos := newGetOmapStep(startAfter, filterPrefix, maxReturn)
+	r.steps = append(r.steps, gos)
+	C.rados_read_op_omap_get_vals2(
+		r.op,
+		gos.cStartAfter,
+		gos.cFilterPrefix,
+		C.uint64_t(gos.maxReturn),
+		&gos.iter,
+		&gos.more,
+		&gos.rval,
+	)
+	return gos
+}
diff --git a/rados/write_op.go b/rados/write_op.go
new file mode 100644
index 0000000..0fab210
--- /dev/null
+++ b/rados/write_op.go
@@ -0,0 +1,179 @@
+package rados
+
+// #cgo LDFLAGS: -lrados
+// #include <errno.h>
+// #include <stdlib.h>
+// #include <rados/librados.h>
+//
+import "C"
+
+import (
+	"unsafe"
+
+	ts "github.com/ceph/go-ceph/internal/timespec"
+)
+
+// Timespec is a public type for the internal C 'struct timespec'
+type Timespec ts.Timespec
+
+// WriteOp manages a set of discrete actions that will be performed together
+// atomically.
+type WriteOp struct {
+	operation
+	op C.rados_write_op_t
+}
+
+// CreateWriteOp returns a newly constructed write operation.
+func CreateWriteOp() *WriteOp {
+	return &WriteOp{
+		op: C.rados_create_write_op(),
+	}
+}
+
+// Release the resources associated with this write operation.
+func (w *WriteOp) Release() {
+	C.rados_release_write_op(w.op)
+	w.op = nil
+	w.free()
+}
+
+func (w WriteOp) operate2(
+	ioctx *IOContext, oid string, mtime *Timespec, flags OperationFlags) error {
+
+	if err := ioctx.validate(); err != nil {
+		return err
+	}
+
+	cOid := C.CString(oid)
+	defer C.free(unsafe.Pointer(cOid))
+	var cMtime *C.struct_timespec
+	if mtime != nil {
+		cMtime = &C.struct_timespec{}
+		ts.CopyToCStruct(
+			ts.Timespec(*mtime),
+			ts.CTimespecPtr(cMtime))
+	}
+
+	ret := C.rados_write_op_operate2(
+		w.op, ioctx.ioctx, cOid, cMtime, C.int(flags))
+	return w.update(writeOp, ret)
+}
+
+// Operate will perform the operation(s).
+func (w *WriteOp) Operate(ioctx *IOContext, oid string, flags OperationFlags) error {
+	return w.operate2(ioctx, oid, nil, flags)
+}
+
+// OperateWithMtime will perform the operation while setting the modification
+// time stamp to the supplied value.
+func (w *WriteOp) OperateWithMtime(
+	ioctx *IOContext, oid string, mtime Timespec, flags OperationFlags) error {
+
+	return w.operate2(ioctx, oid, &mtime, flags)
+}
+
+func (w *WriteOp) operateCompat(ioctx *IOContext, oid string) error {
+	switch err := w.Operate(ioctx, oid, OperationNoFlag).(type) {
+	case nil:
+		return nil
+	case OperationError:
+		return err.OpError
+	default:
+		return err
+	}
+}
+
+// Create a rados object.
+func (w *WriteOp) Create(exclusive CreateOption) {
+	// category, the 3rd param, is deprecated and has no effect so we do not
+	// implement it in go-ceph
+	C.rados_write_op_create(w.op, C.int(exclusive), nil)
+}
+
+//  SetOmap appends the map `pairs` to the omap `oid`.
+func (w *WriteOp) SetOmap(pairs map[string][]byte) {
+	sos := newSetOmapStep(pairs)
+	w.steps = append(w.steps, sos)
+	C.rados_write_op_omap_set(
+		w.op,
+		sos.cKeys,
+		sos.cValues,
+		sos.cLengths,
+		sos.cNum)
+}
+
+// RmOmapKeys removes the specified `keys` from the omap `oid`.
+func (w *WriteOp) RmOmapKeys(keys []string) {
+	roks := newRemoveOmapKeysStep(keys)
+	w.steps = append(w.steps, roks)
+	C.rados_write_op_omap_rm_keys(
+		w.op,
+		roks.cKeys,
+		roks.cNum)
+}
+
+// CleanOmap clears the omap `oid`.
+func (w *WriteOp) CleanOmap() {
+	C.rados_write_op_omap_clear(w.op)
+}
+
+// AssertExists assures the object targeted by the write op exists.
+//
+// Implements:
+//  void rados_write_op_assert_exists(rados_write_op_t write_op);
+func (w *WriteOp) AssertExists() {
+	C.rados_write_op_assert_exists(w.op)
+}
+
+// Write a given byte slice at the supplied offset.
+//
+// Implements:
+//  void rados_write_op_write(rados_write_op_t write_op,
+//                                       const char *buffer,
+//                                       size_t len,
+//                                       uint64_t offset);
+func (w *WriteOp) Write(b []byte, offset uint64) {
+	oe := newWriteStep(b, 0, offset)
+	w.steps = append(w.steps, oe)
+	C.rados_write_op_write(
+		w.op,
+		oe.cBuffer,
+		oe.cDataLen,
+		oe.cOffset)
+}
+
+// WriteFull writes a given byte slice as the whole object,
+// atomically replacing it.
+//
+// Implements:
+//  void rados_write_op_write_full(rados_write_op_t write_op,
+//                                 const char *buffer,
+//                                 size_t len);
+func (w *WriteOp) WriteFull(b []byte) {
+	oe := newWriteStep(b, 0, 0)
+	w.steps = append(w.steps, oe)
+	C.rados_write_op_write_full(
+		w.op,
+		oe.cBuffer,
+		oe.cDataLen)
+}
+
+// WriteSame write a given byte slice to the object multiple times, until
+// writeLen is satisfied.
+//
+// Implements:
+//  void rados_write_op_writesame(rados_write_op_t write_op,
+//                                const char *buffer,
+//                                size_t data_len,
+//                                size_t write_len,
+//                                uint64_t offset);
+func (w *WriteOp) WriteSame(b []byte, writeLen, offset uint64) {
+	oe := newWriteStep(b, writeLen, offset)
+	w.steps = append(w.steps, oe)
+	C.rados_write_op_writesame(
+		w.op,
+		oe.cBuffer,
+		oe.cDataLen,
+		oe.cWriteLen,
+		oe.cOffset)
+}
diff --git a/rados/write_step.go b/rados/write_step.go
new file mode 100644
index 0000000..7ece03f
--- /dev/null
+++ b/rados/write_step.go
@@ -0,0 +1,33 @@
+package rados
+
+// #include <stdint.h>
+import "C"
+
+import (
+	"unsafe"
+)
+
+type writeStep struct {
+	withoutUpdate
+	withoutFree
+	// the c pointer utilizes the Go byteslice data and no free is needed
+
+	// inputs:
+	b []byte
+
+	// arguments:
+	cBuffer   *C.char
+	cDataLen  C.size_t
+	cWriteLen C.size_t
+	cOffset   C.uint64_t
+}
+
+func newWriteStep(b []byte, writeLen, offset uint64) *writeStep {
+	return &writeStep{
+		b:         b,
+		cBuffer:   (*C.char)(unsafe.Pointer(&b[0])),
+		cDataLen:  C.size_t(len(b)),
+		cWriteLen: C.size_t(writeLen),
+		cOffset:   C.uint64_t(offset),
+	}
+}


Test patch:
diff --git a/rados/operation_test.go b/rados/operation_test.go
new file mode 100644
index 0000000..67c6cc8
--- /dev/null
+++ b/rados/operation_test.go
@@ -0,0 +1,119 @@
+package rados
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOperationError(t *testing.T) {
+	oe := OperationError{
+		kind:    readOp,
+		OpError: fmt.Errorf("bad mojo %v", 77),
+		StepErrors: map[int]error{
+			1: fmt.Errorf("limit exceeded"),
+			3: fmt.Errorf("weirdness"),
+		},
+	}
+	assert.Error(t, oe)
+	estr := oe.Error()
+	assert.Contains(t, estr, "read operation error")
+	assert.Contains(t, estr, "op=bad mojo 77")
+	assert.Contains(t, estr, "Step#1=limit exceeded")
+	assert.Contains(t, estr, "Step#3=weirdness")
+
+	oe = OperationError{
+		kind: writeOp,
+		StepErrors: map[int]error{
+			0: fmt.Errorf("unlucky"),
+		},
+	}
+	assert.Error(t, oe)
+	estr = oe.Error()
+	assert.Contains(t, estr, "write operation error")
+	assert.NotContains(t, estr, "op=")
+	assert.Contains(t, estr, "Step#0=unlucky")
+}
+
+type fooStep struct {
+	updateCount int
+	freeCount   int
+	failMe      error
+}
+
+func (fs *fooStep) update() error {
+	fs.updateCount++
+	return fs.failMe
+}
+
+func (fs *fooStep) free() {
+	fs.freeCount++
+}
+
+func TestOpStepFinalizer(t *testing.T) {
+	// this confirms that given a valid op step the finalizer helper
+	// function calls the free method of the interface
+	fs := &fooStep{}
+	opStepFinalizer(fs)
+	opStepFinalizer(fs)
+	assert.Equal(t, 2, fs.freeCount)
+
+	// this should not panic
+	opStepFinalizer(nil)
+}
+
+func TestOperationType(t *testing.T) {
+	t.Run("stepErrors", func(t *testing.T) {
+		o := &operation{}
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{failMe: fmt.Errorf("yow")})
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{failMe: fmt.Errorf("ouch")})
+
+		err := o.update(readOp, 0)
+		if assert.Error(t, err) {
+			oe := err.(OperationError)
+			assert.NoError(t, oe.OpError)
+			assert.Len(t, oe.StepErrors, 2)
+			assert.Contains(t, oe.StepErrors, 1)
+			assert.Contains(t, oe.StepErrors, 4)
+		}
+	})
+
+	t.Run("opError", func(t *testing.T) {
+		o := &operation{}
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{})
+
+		err := o.update(readOp, 5)
+		if assert.Error(t, err) {
+			oe := err.(OperationError)
+			assert.Error(t, oe.OpError)
+			assert.Len(t, oe.StepErrors, 0)
+		}
+	})
+
+	t.Run("noErrors", func(t *testing.T) {
+		o := &operation{}
+		o.steps = append(o.steps, &fooStep{})
+		o.steps = append(o.steps, &fooStep{})
+		err := o.update(readOp, 0)
+		assert.NoError(t, err)
+		x := 0
+		for i := range o.steps {
+			x += o.steps[i].(*fooStep).updateCount
+		}
+		assert.Equal(t, 2, x)
+
+		o.free()
+		x = 0
+		for i := range o.steps {
+			x += o.steps[i].(*fooStep).updateCount
+		}
+		assert.Equal(t, 2, x)
+	})
+}
diff --git a/rados/read_op_test.go b/rados/read_op_test.go
new file mode 100644
index 0000000..e9e319f
--- /dev/null
+++ b/rados/read_op_test.go
@@ -0,0 +1,144 @@
+package rados
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func (suite *RadosTestSuite) TestReadOpAssertExists() {
+	suite.SetupConnection()
+	oid := "TestReadOpAssertExists"
+
+	wrop := CreateWriteOp()
+	defer wrop.Release()
+	wrop.Create(CreateIdempotent)
+	err := wrop.Operate(suite.ioctx, oid, OperationNoFlag)
+	assert.NoError(suite.T(), err)
+
+	op := CreateReadOp()
+	defer op.Release()
+	op.AssertExists()
+	err = op.Operate(suite.ioctx, oid, OperationNoFlag)
+	assert.NoError(suite.T(), err)
+
+	op2 := CreateReadOp()
+	defer op2.Release()
+	op2.AssertExists()
+	err = op2.Operate(suite.ioctx, oid+"junk", OperationNoFlag)
+	assert.Error(suite.T(), err)
+
+	// ensure a nil ioctx triggers a panic
+	assert.Panics(suite.T(), func() {
+		op2.Operate(nil, "foo", OperationNoFlag)
+	})
+}
+
+func getAllMap(gos *GetOmapStep) map[string][]byte {
+	r := make(map[string][]byte)
+	for {
+		kv, err := gos.Next()
+		if err != nil {
+			panic(err)
+		}
+		if kv == nil {
+			break
+		}
+		r[kv.Key] = kv.Value
+	}
+	return r
+}
+
+func (suite *RadosTestSuite) TestReadOpGetOmapValues() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+	oid := "TestReadOpGetOmapValues"
+
+	wrop := CreateWriteOp()
+	defer wrop.Release()
+	wrop.Create(CreateIdempotent)
+	wrop.SetOmap(map[string][]byte{
+		"tos.captain":       []byte("Kirk"),
+		"tos.first-officer": []byte("Spock"),
+		"tos.doctor":        []byte("McCoy"),
+		"tng.captain":       []byte("Picard"),
+		"tng.first-officer": []byte("Riker"),
+		"tng.doctor":        []byte("Crusher"),
+		"random.value":      []byte("foobar"),
+		"no.value":          []byte(""),
+	})
+	err := wrop.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	suite.T().Run("simple", func(t *testing.T) {
+		ta := assert.New(t)
+		op := CreateReadOp()
+		defer op.Release()
+		op.AssertExists()
+		gos := op.GetOmapValues("", "", 16)
+		err = op.Operate(suite.ioctx, oid, OperationNoFlag)
+		ta.NoError(err)
+
+		omap := getAllMap(gos)
+		ta.Len(omap, 8)
+		ta.Contains(omap, "tos.captain")
+		ta.Contains(omap, "tng.captain")
+		ta.False(gos.More())
+	})
+
+	suite.T().Run("twoIterations", func(t *testing.T) {
+		// test two iterations over different subsets of omap keys
+		ta := assert.New(t)
+		op := CreateReadOp()
+		defer op.Release()
+		op.AssertExists()
+		gos1 := op.GetOmapValues("", "tos", 16)
+		gos2 := op.GetOmapValues("", "tng", 16)
+		err = op.Operate(suite.ioctx, oid, OperationNoFlag)
+		ta.NoError(err)
+
+		omap1 := getAllMap(gos1)
+		ta.Len(omap1, 3)
+		ta.Contains(omap1, "tos.captain")
+		ta.Contains(omap1, "tos.first-officer")
+		ta.Contains(omap1, "tos.doctor")
+		omap2 := getAllMap(gos2)
+		ta.Len(omap2, 3)
+		ta.Contains(omap2, "tng.captain")
+		ta.Contains(omap2, "tng.first-officer")
+		ta.Contains(omap2, "tng.doctor")
+	})
+
+	suite.T().Run("checkForMore", func(t *testing.T) {
+		// test two iterations over different subsets of omap keys
+		ta := assert.New(t)
+		op := CreateReadOp()
+		defer op.Release()
+		op.AssertExists()
+		gos := op.GetOmapValues("", "", 6)
+		err = op.Operate(suite.ioctx, oid, OperationNoFlag)
+		ta.NoError(err)
+
+		omap1 := getAllMap(gos)
+		ta.Len(omap1, 6)
+		ta.True(gos.More())
+	})
+
+	suite.T().Run("iterateTooEarly", func(t *testing.T) {
+		// test two iterations over different subsets of omap keys
+		ta := assert.New(t)
+		op := CreateReadOp()
+		defer op.Release()
+		op.AssertExists()
+		gos := op.GetOmapValues("", "", 6)
+		_, err := gos.Next()
+		ta.Error(err)
+		ta.Equal(ErrOperationIncomplete, err)
+	})
+}
+
+func TestReadOpInvalid(t *testing.T) {
+	r := &ReadOp{}
+	err := r.Operate(&IOContext{}, "foo", 0)
+	assert.Error(t, err)
+}
diff --git a/rados/write_op_test.go b/rados/write_op_test.go
new file mode 100644
index 0000000..3a58a58
--- /dev/null
+++ b/rados/write_op_test.go
@@ -0,0 +1,247 @@
+package rados
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+// timeStamp generates a dummy Timespec value.
+func timeStamp() Timespec {
+	// Future TODO (maybe?) - vary the value?
+	return Timespec{342334800, 0}
+}
+
+func (suite *RadosTestSuite) TestWriteOpCreate() {
+	suite.SetupConnection()
+
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	err := op.Operate(suite.ioctx, "TestWriteOpCreate", OperationNoFlag)
+	assert.NoError(suite.T(), err)
+
+	op2 := CreateWriteOp()
+	defer op2.Release()
+	op2.Create(CreateExclusive)
+	err = op2.Operate(suite.ioctx, "TestWriteOpCreate", OperationNoFlag)
+	assert.Error(suite.T(), err)
+
+	// ensure a nil ioctx triggers a panic
+	assert.Panics(suite.T(), func() {
+		op := CreateWriteOp()
+		defer op.Release()
+		op.Operate(nil, "foo", OperationNoFlag)
+	})
+}
+
+func (suite *RadosTestSuite) TestWriteOpCreateWithTimestamp() {
+	suite.SetupConnection()
+
+	oid := "TestWriteOpCreateWithTimestamp"
+	gts := timeStamp()
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	err := op.OperateWithMtime(suite.ioctx, oid, gts, OperationNoFlag)
+	assert.NoError(suite.T(), err)
+
+	s, err := suite.ioctx.Stat(oid)
+	assert.NoError(suite.T(), err)
+	statTime := s.ModTime.Unix()
+	assert.Equal(suite.T(), gts.Sec, statTime)
+}
+
+func (suite *RadosTestSuite) TestWriteOpSetOmap() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpSetOmap"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	op.SetOmap(map[string][]byte{
+		"alice":   []byte("car"),
+		"boss":    []byte("office"),
+		"catbert": []byte("dungeon"),
+	})
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	// the 2nd set of omap values should not be applied because
+	// the Create will fail to exclusively make the object
+	op2 := CreateWriteOp()
+	defer op2.Release()
+	op.Create(CreateExclusive)
+	op.SetOmap(map[string][]byte{
+		"alice":   []byte("home"),
+		"boss":    []byte("golf course"),
+		"catbert": []byte("dungeon"),
+	})
+	err = op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.Error(err)
+
+	fetched, err := suite.ioctx.GetOmapValues(oid, "", "", 10)
+	ta.NoError(err)
+	ta.Equal("car", string(fetched["alice"]))
+	ta.Equal("office", string(fetched["boss"]))
+}
+
+func (suite *RadosTestSuite) TestWriteOpRmOmapKeys() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpRmOmapKeys"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	op.SetOmap(map[string][]byte{
+		"alice":   []byte("car"),
+		"boss":    []byte("office"),
+		"catbert": []byte("dungeon"),
+	})
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	op2 := CreateWriteOp()
+	defer op2.Release()
+	op2.Create(CreateIdempotent)
+	op2.SetOmap(map[string][]byte{
+		"dogbert": []byte("lab"),
+	})
+	op2.RmOmapKeys([]string{"catbert"})
+	err = op2.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	fetched, err := suite.ioctx.GetOmapValues(oid, "", "", 10)
+	ta.NoError(err)
+	ta.Len(fetched, 3)
+	ta.Equal("car", string(fetched["alice"]))
+	ta.Equal("office", string(fetched["boss"]))
+	ta.Equal("lab", string(fetched["dogbert"]))
+	ta.NotContains(fetched, "catbert")
+}
+
+func (suite *RadosTestSuite) TestWriteOpCleanOmap() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpCleanOmap"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	op.SetOmap(map[string][]byte{
+		"alice":   []byte("car"),
+		"boss":    []byte("office"),
+		"catbert": []byte("dungeon"),
+	})
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	// this test simulates wanting to start a fresh new set of
+	// omap keys, atomically clearing and setting a new key & value.
+	op2 := CreateWriteOp()
+	defer op2.Release()
+	op2.Create(CreateIdempotent)
+	op2.CleanOmap()
+	op2.SetOmap(map[string][]byte{
+		"dogbert": []byte("lab"),
+	})
+	op2.RmOmapKeys([]string{"catbert"})
+	err = op2.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	fetched, err := suite.ioctx.GetOmapValues(oid, "", "", 10)
+	ta.NoError(err)
+	ta.Len(fetched, 1)
+	ta.Equal("lab", string(fetched["dogbert"]))
+}
+
+func (suite *RadosTestSuite) TestWriteOpAssertExists() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpRmOmapKeys"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	op2 := CreateWriteOp()
+	defer op2.Release()
+	op2.AssertExists()
+	op2.CleanOmap()
+	err = op2.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	op3 := CreateWriteOp()
+	defer op3.Release()
+	op3.AssertExists()
+	op3.CleanOmap()
+	err = op3.Operate(suite.ioctx, oid+"dne", OperationNoFlag)
+	ta.Error(err)
+}
+
+func (suite *RadosTestSuite) TestWriteOpWrite() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpWrite"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	op.Write([]byte("go-go-gadget"), 0)
+	op.Write([]byte("ceph project!"), 6)
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	d := make([]byte, 32)
+	l, err := suite.ioctx.Read(oid, d, 0)
+	ta.NoError(err)
+	ta.Equal("go-go-ceph project!", string(d[:l]))
+}
+
+func (suite *RadosTestSuite) TestWriteOpWriteFull() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpWriteFull"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	op.WriteFull([]byte("one, two"))
+	op.WriteFull([]byte("buckle my shoe"))
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	d := make([]byte, 32)
+	l, err := suite.ioctx.Read(oid, d, 0)
+	ta.NoError(err)
+	ta.Equal("buckle my shoe", string(d[:l]))
+}
+
+func (suite *RadosTestSuite) TestWriteOpWriteSame() {
+	suite.SetupConnection()
+	ta := assert.New(suite.T())
+
+	oid := "TestWriteOpWriteSame"
+	op := CreateWriteOp()
+	defer op.Release()
+	op.Create(CreateIdempotent)
+	op.WriteSame([]byte("repetition "), 44, 0)
+	op.Write([]byte("is fun"), 44)
+	err := op.Operate(suite.ioctx, oid, OperationNoFlag)
+	ta.NoError(err)
+
+	d := make([]byte, 64)
+	l, err := suite.ioctx.Read(oid, d, 0)
+	ta.NoError(err)
+	ta.Equal("repetition repetition repetition repetition is fun", string(d[:l]))
+}
+
+func TestWriteOpInvalid(t *testing.T) {
+	r := &WriteOp{}
+	err := r.Operate(&IOContext{}, "foo", 0)
+	assert.Error(t, err)
+}