Repository: hamba/avro
Pull request #76 (2021-01-22 23:35:57)

ocf: fix header schema to avoid data race
I use this excellent avro library because this is three times faster than linkedin/goavro in my work load.
I found that data-race occurred in multi thread program when I used this library.
That is because `ocf.go` shares `HeaderSchema` instance in `github.com/hamba/avro.(*fingerprinter).Fingerprint()` which is not thread safe.

At first, I attempted to fix `fingerprinter.Fingerprint()` to make it thread safe, but it ended up with less performance.
So, I fix `ocf.go` to avoid sharing a `HeaderSchema` instance.

I write a simple data-race example.
https://github.com/getumen/avro-datarace-example/blob/main/main.go
```
[yoshihiro@MBP] ( ﾟ∀ﾟ)・∵. ~/github/avro-datarace-example % go run -race ./main.go                                                                                                                                                                        (git)-[main]
==================
WARNING: DATA RACE
Read at 0x00c0000dc2d0 by goroutine 10:
  github.com/hamba/avro.(*fingerprinter).Fingerprint()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/schema.go:221 +0x16c
  github.com/hamba/avro.(*RecordSchema).Fingerprint()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/schema.go:407 +0x5c
  github.com/hamba/avro.(*Reader).ReadVal()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/codec.go:42 +0xc3
  github.com/hamba/avro/ocf.NewDecoder()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/ocf/ocf.go:63 +0x129
  main.main.func1()
      /Users/yoshihiro/github/avro-datarace-example/main.go:31 +0xdc

Previous write at 0x00c0000dc2d0 by goroutine 8:
  github.com/hamba/avro.(*fingerprinter).Fingerprint()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/schema.go:224 +0xf4
  github.com/hamba/avro.(*RecordSchema).Fingerprint()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/schema.go:407 +0x5c
  github.com/hamba/avro.(*Reader).ReadVal()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/codec.go:42 +0xc3
  github.com/hamba/avro/ocf.NewDecoder()
      /Users/yoshihiro/go/1.15.3/pkg/mod/github.com/hamba/avro@v1.5.1/ocf/ocf.go:63 +0x129
  main.main.func1()
      /Users/yoshihiro/github/avro-datarace-example/main.go:31 +0xdc

Goroutine 10 (running) created at:
  main.main()
      /Users/yoshihiro/github/avro-datarace-example/main.go:27 +0x126

Goroutine 8 (running) created at:
  main.main()
      /Users/yoshihiro/github/avro-datarace-example/main.go:27 +0x126
==================
```

nrwiersma: Hi,

Thanks for pointing this out. It seems the entire fingerprinter is not safe for concurrent use unlike the rest of schema which is designed for concurrent use.

While I agree that fixing this in `OCF` will fix the lib issue, I feel the problem will crop its head up again at some point if it is not solved at the source. I will run some benchmarks to see that the lowest cost fix could be. My guess is that a RWMutex will be best which should cost an extra ~25ns by my last bench, but I also wonder if there is a lockless way to resolve this.
getumen: Thank you for your response!

I also think RWMutex is a good choice.
When I use the below code, I found no difference between this PR and the below implementation in my workload.
However, this code may decrease the performance of this library that has an advantage in the performance.
So I proposed this modification which did not decrease the performance.

```
type fingerprinter struct {
	sync.RWMutex
	fingerprint [32]byte
	cache       map[FingerprintType][]byte
}

// Fingerprint returns the SHA256 fingerprint of the schema.
func (f *fingerprinter) Fingerprint(stringer fmt.Stringer) [32]byte {
	f.RLock()
	if f.fingerprint != emptyFgpt {
		defer f.RUnlock()
		return f.fingerprint
	}
	f.RUnlock()
	f.Lock()
	defer f.Unlock()
	f.fingerprint = sha256.Sum256([]byte(stringer.String()))
	return f.fingerprint
}

```
getumen: I think initialized flag protected by atomic package may be faster because an atomic operation is one machine instruction.
nrwiersma: I agree. Looking at the benchmarks I saw a 3x decrease with RWMutex which is too much in my opinion. I am also exploring the `atomic` package as a solution, but I am looking into `atomic.StorePointer` to see if this will work. Atomic booleans could also provide a good choice, but will likely be more complex than dealing with atomic pointers. 

I will explore more today and hopefully we will have some final benched choices by the end of the weekend.
nrwiersma: ```
type fingerprinter struct {
	fingerprint atomic.Value // [32]byte
	cache concurrent.Map //map[FingerprintType][]byte
}

// Fingerprint returns the SHA256 fingerprint of the schema.
func (f *fingerprinter) Fingerprint(stringer fmt.Stringer) [32]byte {
	if v := f.fingerprint.Load(); v != nil {
		return v.([32]byte)
	}

	fingerprint := sha256.Sum256([]byte(stringer.String()))
	f.fingerprint.Store(fingerprint)
	return fingerprint
}

// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
func (f *fingerprinter) FingerprintUsing(typ FingerprintType, stringer fmt.Stringer) ([]byte, error) {
	if v, ok := f.cache.Load(typ); ok {
		return v.([]byte), nil
	}

	h, ok := fingerprinters[typ]
	if !ok {
		return nil, fmt.Errorf("avro: unknown fingerprint algorithm %s", typ)
	}

	h.Reset()
	_, _ = h.Write([]byte(stringer.String()))
	fingerprint := h.Sum(make([]byte, 0, h.Size()))
	f.cache.Store(typ, fingerprint)
	return fingerprint, nil
}
```

This solution for me provides atomic safety with no performance decrease (I am ignoring the performance drop in `FingerprintUsing`). Can you perhaps confirm on your side?
getumen: I will try it.
I guess `concurrent.Map` is `sync.Map`. Is it right?

getumen: I modify `fingerprinter` concurrent safe.
https://github.com/hamba/avro/pull/76/commits/d2a903229bce5abbd573a6745144d0915ada7047

Also, I executed my workload and benchmark in https://github.com/nrwiersma/avro-benchmarks.
In my workload, I found no difference between 'not concurrent-safe-implementation' and this implementation.
In benchmark(https://github.com/nrwiersma/avro-benchmarks), the result looks good.
```
[yoshihiro@MBP] ( ﾟ∀ﾟ)・∵. ~/github/avro-benchmarks % go test -bench . -benchmem                                                                                                                                                        (git)-[feature-local-avro-lib]
goos: darwin
goarch: amd64
pkg: github.com/nrwiersma/avro-benchmarks
BenchmarkGoAvroDecode-4     	  323404	      3657 ns/op	     442 B/op	      27 allocs/op
BenchmarkGoAvroEncode-4     	  268454	      4410 ns/op	     871 B/op	      63 allocs/op
BenchmarkHambaDecode-4      	 2092245	       577 ns/op	      64 B/op	       4 allocs/op
BenchmarkHambaEncode-4      	 2364170	       498 ns/op	     112 B/op	       1 allocs/op
BenchmarkLinkedinDecode-4   	  459288	      2263 ns/op	    1728 B/op	      35 allocs/op
BenchmarkLinkedinEncode-4   	 1436437	       912 ns/op	     248 B/op	       5 allocs/op
PASS
ok  	github.com/nrwiersma/avro-benchmarks	10.184s
```

Patch:
diff --git a/schema.go b/schema.go
index 513bbfd..6760aa4 100644
--- a/schema.go
+++ b/schema.go
@@ -8,15 +8,13 @@ import (
 	"hash"
 	"strconv"
 	"strings"
+	"sync/atomic"
 
 	"github.com/hamba/avro/pkg/crc64"
 	"github.com/modern-go/concurrent"
 )
 
-var (
-	emptyFgpt   = [32]byte{}
-	nullDefault = struct{}{}
-)
+var nullDefault = struct{}{}
 
 // Type is a schema type.
 type Type string
@@ -211,28 +209,25 @@ func (n name) FullName() string {
 }
 
 type fingerprinter struct {
-	fingerprint [32]byte
-	cache       map[FingerprintType][]byte
+	fingerprint atomic.Value   // [32]byte
+	cache       concurrent.Map // map[FingerprintType][]byte
 }
 
 // Fingerprint returns the SHA256 fingerprint of the schema.
 func (f *fingerprinter) Fingerprint(stringer fmt.Stringer) [32]byte {
-	if f.fingerprint != emptyFgpt {
-		return f.fingerprint
+	if v := f.fingerprint.Load(); v != nil {
+		return v.([32]byte)
 	}
 
-	f.fingerprint = sha256.Sum256([]byte(stringer.String()))
-	return f.fingerprint
+	fingerprint := sha256.Sum256([]byte(stringer.String()))
+	f.fingerprint.Store(fingerprint)
+	return fingerprint
 }
 
 // FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
 func (f *fingerprinter) FingerprintUsing(typ FingerprintType, stringer fmt.Stringer) ([]byte, error) {
-	if f.cache == nil {
-		f.cache = map[FingerprintType][]byte{}
-	}
-
-	if b, ok := f.cache[typ]; ok {
-		return b, nil
+	if v, ok := f.cache.Load(typ); ok {
+		return v.([]byte), nil
 	}
 
 	h, ok := fingerprinters[typ]
@@ -243,7 +238,7 @@ func (f *fingerprinter) FingerprintUsing(typ FingerprintType, stringer fmt.Strin
 	h.Reset()
 	_, _ = h.Write([]byte(stringer.String()))
 	fingerprint := h.Sum(make([]byte, 0, h.Size()))
-	f.cache[typ] = fingerprint
+	f.cache.Store(typ, fingerprint)
 	return fingerprint, nil
 }
 


Test patch:
diff --git a/schema_internal_test.go b/schema_internal_test.go
index a5b79b3..ab1bcf5 100644
--- a/schema_internal_test.go
+++ b/schema_internal_test.go
@@ -333,6 +333,8 @@ func TestSchema_FingerprintUsingCaches(t *testing.T) {
 
 	got, _ := schema.FingerprintUsing(CRC64Avro)
 
-	assert.Equal(t, want, schema.cache[CRC64Avro])
+	value, ok := schema.cache.Load(CRC64Avro)
+	assert.True(t, ok)
+	assert.Equal(t, want, value)
 	assert.Equal(t, want, got)
 }