Repository: ceph/go-ceph
Pull request #320 (2020-07-01 17:08:09)

rbd: Add UpdateWatch, Unwatch, and rbd metadata change watch callback support
These patches add support for setting watches on rbd images via a new UpdateWatch function implementing rbd_update_watch as well as Unwatch implementing rbd_update_unwatch, to remove the watch. Tests are included, along with an extensive test case using watch callbacks to feed a go channel. This test exercises multiples watches at once as well as interaction between C and Go code.


## 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: I've been banging my head against the wall all day trying to figure out what kind of race condition in my new test case was causing the ci to fail. The tests run fine locally. Just recently, I stepped away from the computer and when I looked at the issue with fresh eyes I noticed that it didn't even seem to be running the tests. Locally, I force refreshed the container image ceph/daemon-base:latest-octopus and tried it locally. Now, with the latest ocotopus image present I locally reproduce the same behavior. It appears that the micro-osd.sh is getting stuck when running ceph-fuse. I need to investigate more but I ought to be able to start making progress again.
ansiwen: Another thing just came to my mind: what guarantees do we have regarding the scheduling of the threads that call the callbacks? Must the callbacks be at least reentrant-safe, in case a callback gets called twice from different threads? This could get tricky... maybe we would need to only offer channels as API for watchers? :thinking: 
ansiwen: After a long phone conversation where @phlogistonjohn was nice enough to patiently explain everything to me, I approve this PR. :slightly_smiling_face: 
phlogistonjohn: > After a long phone conversation where @phlogistonjohn was nice enough to patiently explain everything to me, I approve this PR. slightly_smiling_face

It was a good conversation, thanks!

Patch:
diff --git a/rbd/callback_shims_mimic.go b/rbd/callback_shims_mimic.go
new file mode 100644
index 0000000..fdf0303
--- /dev/null
+++ b/rbd/callback_shims_mimic.go
@@ -0,0 +1,15 @@
+// +build !luminous
+
+package rbd
+
+/*
+
+#include <rbd/librbd.h>
+
+extern void imageWatchCallback(int index);
+
+void callWatchCallback(int index) {
+	imageWatchCallback(index);
+}
+*/
+import "C"
diff --git a/rbd/watchers_mimic.go b/rbd/watchers_mimic.go
index 198292a..54d5dfd 100644
--- a/rbd/watchers_mimic.go
+++ b/rbd/watchers_mimic.go
@@ -4,12 +4,29 @@
 
 package rbd
 
-// #cgo LDFLAGS: -lrbd
-// #include <errno.h>
-// #include <rbd/librbd.h>
+/*
+#cgo LDFLAGS: -lrbd
+#include <rbd/librbd.h>
+
+extern int callWatchCallback(int index);
+
+// cgo has trouble converting the types of the callback and data arg defined in
+// librbd header. It wants the callback function to be a byte pointer and
+// the arg to be a pointer, which is pretty much the opposite of what we
+// actually want. This shim exists to help coerce the auto-type-conversion
+// to do the right thing for us.
+static inline int wrap_rbd_update_watch(
+			rbd_image_t image,
+			uint64_t *handle,
+			void *watch_cb,
+			uintptr_t arg) {
+	return rbd_update_watch(image, handle, watch_cb, (void*)arg);
+}
+*/
 import "C"
 
 import (
+	"github.com/ceph/go-ceph/internal/callbacks"
 	"github.com/ceph/go-ceph/internal/retry"
 )
 
@@ -59,3 +76,77 @@ func (image *Image) ListWatchers() ([]ImageWatcher, error) {
 	}
 	return imageWatchers, nil
 }
+
+// watchCallbacks tracks the active callbacks for rbd watches
+var watchCallbacks = callbacks.New()
+
+// WatchCallback defines the function signature needed for the UpdateWatch
+// callback.
+type WatchCallback func(interface{})
+
+type watchInstance struct {
+	callback WatchCallback
+	data     interface{}
+}
+
+// Watch represents an ongoing image metadata watch.
+type Watch struct {
+	image   *Image
+	wi      watchInstance
+	handle  C.uint64_t
+	cbIndex int
+}
+
+// UpdateWatch updates the image object to watch metadata changes to the
+// image, returning a Watch object.
+//
+// Implements:
+//  int rbd_update_watch(rbd_image_t image, uint64_t *handle,
+//                       rbd_update_callback_t watch_cb, void *arg);
+func (image *Image) UpdateWatch(cb WatchCallback, data interface{}) (*Watch, error) {
+	if err := image.validate(imageIsOpen); err != nil {
+		return nil, err
+	}
+	wi := watchInstance{
+		callback: cb,
+		data:     data,
+	}
+	w := &Watch{
+		image:   image,
+		wi:      wi,
+		cbIndex: watchCallbacks.Add(wi),
+	}
+
+	ret := C.wrap_rbd_update_watch(
+		image.image,
+		&w.handle,
+		C.callWatchCallback,
+		C.uintptr_t(w.cbIndex))
+	if ret != 0 {
+		return nil, getError(ret)
+	}
+	return w, nil
+}
+
+// Unwatch un-registers the image watch.
+//
+// Implements:
+//  int rbd_update_unwatch(rbd_image_t image, uint64_t handle);
+func (w *Watch) Unwatch() error {
+	if w.image == nil {
+		return ErrImageNotOpen
+	}
+	if err := w.image.validate(imageIsOpen); err != nil {
+		return err
+	}
+	ret := C.rbd_update_unwatch(w.image.image, w.handle)
+	watchCallbacks.Remove(w.cbIndex)
+	return getError(ret)
+}
+
+//export imageWatchCallback
+func imageWatchCallback(index C.int) {
+	v := watchCallbacks.Lookup(int(index))
+	wi := v.(watchInstance)
+	wi.callback(wi.data)
+}


Test patch:
diff --git a/rbd/watchers_mimic_test.go b/rbd/watchers_mimic_test.go
index 451cfef..67fc597 100644
--- a/rbd/watchers_mimic_test.go
+++ b/rbd/watchers_mimic_test.go
@@ -5,7 +5,9 @@
 package rbd
 
 import (
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -115,3 +117,183 @@ func TestListWatchers(t *testing.T) {
 		assert.Equal(t, 0, len(watchers))
 	})
 }
+
+func TestWatch(t *testing.T) {
+	conn := radosConnect(t)
+	require.NotNil(t, conn)
+	defer conn.Shutdown()
+
+	poolname := GetUUID()
+	err := conn.MakePool(poolname)
+	require.NoError(t, err)
+	defer conn.DeletePool(poolname)
+
+	ioctx, err := conn.OpenIOContext(poolname)
+	require.NoError(t, err)
+	defer ioctx.Destroy()
+
+	startSize := uint64(1 << 21)
+	name := GetUUID()
+	options := NewRbdImageOptions()
+	err = CreateImage(ioctx, name, startSize, options)
+	require.NoError(t, err)
+	defer func() { assert.NoError(t, RemoveImage(ioctx, name)) }()
+
+	t.Run("imageNotOpen", func(t *testing.T) {
+		image, err := OpenImageReadOnly(ioctx, name, NoSnapshot)
+		require.NoError(t, err)
+		require.NotNil(t, image)
+
+		err = image.Close()
+		require.NoError(t, err)
+
+		_, err = image.UpdateWatch(func(d interface{}) {
+		}, nil)
+		assert.Equal(t, ErrImageNotOpen, err)
+	})
+
+	t.Run("simpleWatch", func(t *testing.T) {
+		image, err := OpenImage(ioctx, name, NoSnapshot)
+		require.NoError(t, err)
+		require.NotNil(t, image)
+
+		defer func() {
+			assert.NoError(t, image.Close())
+		}()
+
+		cc := 0
+		w, err := image.UpdateWatch(func(d interface{}) {
+			cc++
+		}, nil)
+		assert.NoError(t, err)
+		defer func() {
+			assert.NoError(t, w.Unwatch())
+		}()
+
+		x := make(chan int)
+		defer close(x)
+		go func() {
+			for i := 0; i < 5; i++ {
+				i1, err := OpenImage(ioctx, name, NoSnapshot)
+				err = i1.Resize(startSize * uint64(1+i))
+				assert.NoError(t, err)
+				err = i1.Close()
+				assert.NoError(t, err)
+				time.Sleep(5 * time.Millisecond)
+			}
+			x <- 0
+		}()
+		<-x
+
+		assert.Equal(t, 5, cc)
+	})
+
+	t.Run("badWatch", func(t *testing.T) {
+		w := &Watch{}
+		err := w.Unwatch()
+		assert.Error(t, err)
+
+		i1, err := OpenImage(ioctx, name, NoSnapshot)
+		assert.NoError(t, err)
+		assert.NoError(t, i1.Close())
+		w.image = i1
+		err = w.Unwatch()
+		assert.Error(t, err)
+	})
+}
+
+func TestWatchMultiChannels(t *testing.T) {
+	conn := radosConnect(t)
+	require.NotNil(t, conn)
+	defer conn.Shutdown()
+
+	poolname := GetUUID()
+	err := conn.MakePool(poolname)
+	require.NoError(t, err)
+	defer conn.DeletePool(poolname)
+
+	ioctx, err := conn.OpenIOContext(poolname)
+	require.NoError(t, err)
+	defer ioctx.Destroy()
+
+	startSize := uint64(1 << 21)
+	images := map[string]uint64{}
+	for i := 0; i < 4; i++ {
+		name := GetUUID()
+		images[name] = startSize
+		options := NewRbdImageOptions()
+		err = CreateImage(ioctx, name, startSize, options)
+		require.NoError(t, err)
+		defer func() {
+			assert.NoError(t, RemoveImage(ioctx, name))
+		}()
+	}
+
+	var wg sync.WaitGroup
+	watchMe := func(n string, mon chan<- string, done <-chan bool) {
+		img, err := OpenImage(ioctx, n, NoSnapshot)
+		require.NoError(t, err)
+		defer func() {
+			assert.NoError(t, img.Close())
+		}()
+		w, err := img.UpdateWatch(func(d interface{}) {
+			mon <- n
+		}, nil)
+		require.NoError(t, err)
+		defer func() {
+			assert.NoError(t, w.Unwatch())
+		}()
+
+		wg.Done() // our watch is set up and active
+		<-done
+	}
+
+	mon := make(chan string)
+	defer close(mon)
+	done := make(chan bool, len(images))
+	defer close(done)
+	wg.Add(len(images))
+	for n := range images {
+		go watchMe(n, mon, done)
+	}
+	wg.Wait()
+
+	x := make(chan bool)
+	go func() {
+		inames := []string{}
+		for n := range images {
+			inames = append(inames, n)
+		}
+		for i := 0; i < 12; i++ {
+			n := inames[i%len(inames)]
+			images[n] *= uint64(2)
+			i1, err := OpenImage(ioctx, n, NoSnapshot)
+			err = i1.Resize(images[n])
+			assert.NoError(t, err)
+			err = i1.Close()
+			assert.NoError(t, err)
+			time.Sleep(5 * time.Millisecond)
+		}
+		time.Sleep(5 * time.Millisecond)
+		for range inames {
+			done <- true
+		}
+		x <- true
+	}()
+
+	ncount := map[string]int{}
+	for ok := true; ok; {
+		select {
+		case n := <-mon:
+			ncount[n]++
+		case <-x:
+			ok = false
+			break
+		}
+	}
+
+	assert.Len(t, ncount, 4)
+	for _, v := range ncount {
+		assert.Equal(t, 3, v)
+	}
+}