fixed concurrency issues in deferredMarshaller

Author: vjc1234

gateway/router/marshal/json/cache.go

@@ -100,25 +100,23 @@ func (m *marshallersCache) loadMarshaller(rType reflect.Type, config *config.IOC
 	return marshaller, nil
 }
 
-func (c *pathCache) loadOrGetMarshaller(rType reflect.Type, config *config.IOConfig, path string, outputPath string, tag *format.Tag, options ...interface{}) (marshaler, error) {
-	value, ok := c.cache.Load(rType)
+func (c *pathCache) loadOrGetMarshaller(rType reflect.Type, cfg *config.IOConfig, path, outPath string, tag *format.Tag, options ...interface{}) (marshaler, error) {
+
+	placeholder := newDeferred()
+	value, ok := c.cache.LoadOrStore(rType, placeholder)
 	if ok {
 		return value.(marshaler), nil
 	}
 
-	// Place a deferred placeholder to break recursive graphs for this path and type.
-	placeholder := &deferredMarshaller{}
-	c.storeMarshaler(rType, placeholder)
-
-	aMarshaler, err := c.getMarshaller(rType, config, path, outputPath, tag, options...)
+	aMarshaller, err := c.getMarshaller(rType, cfg, path, outPath, tag, options...)
 	if err != nil {
+		placeholder.fail(err)                        // unblock anyone holding the promise
+		c.cache.CompareAndDelete(rType, placeholder) // allow a clean retry later
 		return nil, err
 	}
 
-	// Swap placeholder with the real marshaller and set target for any users that captured it.
-	placeholder.setTarget(aMarshaler)
-	c.storeMarshaler(rType, aMarshaler)
-	return aMarshaler, nil
+	placeholder.setTarget(aMarshaller) // resolve success
+	return aMarshaller, nil
 }
 
 func (c *pathCache) getMarshaller(rType reflect.Type, config *config.IOConfig, path string, outputPath string, tag *format.Tag, options ...interface{}) (marshaler, error) {

gateway/router/marshal/json/marshaller_deferred.go

@@ -2,30 +2,56 @@ package json
 
 import (
 	"fmt"
-	"github.com/francoispqt/gojay"
 	"unsafe"
+
+	"github.com/francoispqt/gojay"
 )
 
 // deferredMarshaller is a placeholder used to break recursive type graphs during construction.
 // It forwards calls to the actual target once it is set.
 type deferredMarshaller struct {
 	target marshaler
+	ready  chan struct{}
+	err    error
+}
+
+func newDeferred() *deferredMarshaller {
+	return &deferredMarshaller{ready: make(chan struct{})}
 }
 
 func (d *deferredMarshaller) setTarget(m marshaler) {
 	d.target = m
+	close(d.ready)
+}
+
+func (d *deferredMarshaller) fail(e error) {
+	d.err = e
+	close(d.ready) // writes to err happen-before any receive on ready
 }
 
-func (d *deferredMarshaller) MarshallObject(ptr unsafe.Pointer, session *MarshallSession) error {
+func (d *deferredMarshaller) resolved() (marshaler, error) {
+	<-d.ready // wait for resolve/fail
+	if d.err != nil {
+		return nil, d.err
+	}
 	if d.target == nil {
-		return fmt.Errorf("marshaller not initialized")
+		return nil, fmt.Errorf("marshaller not initialized")
 	}
-	return d.target.MarshallObject(ptr, session)
+	return d.target, nil
 }
 
-func (d *deferredMarshaller) UnmarshallObject(pointer unsafe.Pointer, decoder *gojay.Decoder, auxiliaryDecoder *gojay.Decoder, session *UnmarshalSession) error {
-	if d.target == nil {
-		return fmt.Errorf("marshaller not initialized")
+func (d *deferredMarshaller) MarshallObject(ptr unsafe.Pointer, s *MarshallSession) error {
+	m, err := d.resolved()
+	if err != nil {
+		return err
+	}
+	return m.MarshallObject(ptr, s)
+}
+
+func (d *deferredMarshaller) UnmarshallObject(p unsafe.Pointer, dec, aux *gojay.Decoder, s *UnmarshalSession) error {
+	m, err := d.resolved()
+	if err != nil {
+		return err
 	}
-	return d.target.UnmarshallObject(pointer, decoder, auxiliaryDecoder, session)
+	return m.UnmarshallObject(p, dec, aux, s)
 }