Refactor metric records logic (#468)

* Refactor metric records logic.

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Fix lint errors

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Fix a bug that we try to readd the old entry instead of a new one.

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Update comments in refcount_mapped.

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Remove the need to use a records list, iterate over the map.

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Fix comments and typos

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Fix more comments

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Clarify tryUnmap comment

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

* Fix one more typo.

Signed-off-by: Bogdan Cristian Drutu <[email protected]>

sdk/metric/alignment_test.go

@@ -12,20 +12,12 @@ import (
 func TestMain(m *testing.M) {
 	fields := []ottest.FieldOffset{
 		{
-			Name:   "record.refcount",
-			Offset: unsafe.Offsetof(record{}.refcount),
+			Name:   "record.refMapped.value",
+			Offset: unsafe.Offsetof(record{}.refMapped.value),
 		},
 		{
-			Name:   "record.collectedEpoch",
-			Offset: unsafe.Offsetof(record{}.collectedEpoch),
-		},
-		{
-			Name:   "record.modifiedEpoch",
-			Offset: unsafe.Offsetof(record{}.modifiedEpoch),
-		},
-		{
-			Name:   "record.reclaim",
-			Offset: unsafe.Offsetof(record{}.reclaim),
+			Name:   "record.modified",
+			Offset: unsafe.Offsetof(record{}.modified),
 		},
 	}
 	if !ottest.Aligned8Byte(fields, os.Stderr) {

sdk/metric/list.go

@@ -14,11 +14,6 @@
 
 package metric
 
-import (
-	"sync/atomic"
-	"unsafe"
-)
-
 func (l *sortedLabels) Len() int {
 	return len(*l)
 }
@@ -30,51 +25,3 @@ func (l *sortedLabels) Swap(i, j int) {
 func (l *sortedLabels) Less(i, j int) bool {
 	return (*l)[i].Key < (*l)[j].Key
 }
-
-func (m *SDK) addPrimary(rec *record) {
-	for {
-		rec.next.primary.store(m.records.primary.load())
-		if atomic.CompareAndSwapPointer(
-			&m.records.primary.ptr,
-			rec.next.primary.ptr,
-			unsafe.Pointer(rec),
-		) {
-			return
-		}
-	}
-}
-
-func (m *SDK) addReclaim(rec *record) {
-	for {
-		rec.next.reclaim.store(m.records.reclaim.load())
-		if atomic.CompareAndSwapPointer(
-			&m.records.reclaim.ptr,
-			rec.next.reclaim.ptr,
-			unsafe.Pointer(rec),
-		) {
-			return
-		}
-	}
-}
-
-func (s *singlePtr) swapNil() *record {
-	for {
-		newValue := unsafe.Pointer(nil)
-		swapped := atomic.LoadPointer(&s.ptr)
-		if atomic.CompareAndSwapPointer(&s.ptr, swapped, newValue) {
-			return (*record)(swapped)
-		}
-	}
-}
-
-func (s *singlePtr) load() *record {
-	return (*record)(atomic.LoadPointer(&s.ptr))
-}
-
-func (s *singlePtr) store(r *record) {
-	atomic.StorePointer(&s.ptr, unsafe.Pointer(r))
-}
-
-func (s *singlePtr) clear() {
-	atomic.StorePointer(&s.ptr, unsafe.Pointer(nil))
-}

sdk/metric/refcount_mapped.go

@@ -0,0 +1,65 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metric
+
+import (
+	"sync/atomic"
+)
+
+// refcountMapped atomically counts the number of references (usages) of an entry
+// while also keeping a state of mapped/unmapped into a different data structure
+// (an external map or list for example).
+//
+// refcountMapped uses an atomic value where the least significant bit is used to
+// keep the state of mapping ('1' is used for unmapped and '0' is for mapped) and
+// the rest of the bits are used for refcounting.
+type refcountMapped struct {
+	// refcount has to be aligned for 64-bit atomic operations.
+	value int64
+}
+
+// ref returns true if the entry is still mapped and increases the
+// reference usages, if unmapped returns false.
+func (rm *refcountMapped) ref() bool {
+	// Check if this entry was marked as unmapped between the moment
+	// we got a reference to it (or will be removed very soon) and here.
+	return atomic.AddInt64(&rm.value, 2)&1 == 0
+}
+
+func (rm *refcountMapped) unref() {
+	atomic.AddInt64(&rm.value, -2)
+}
+
+// inUse returns true if there is a reference to the entry and it is mapped.
+func (rm *refcountMapped) inUse() bool {
+	val := atomic.LoadInt64(&rm.value)
+	return val >= 2 && val&1 == 0
+}
+
+// tryUnmap flips the mapped bit to "unmapped" state and returns true if both of the
+// following conditions are true upon entry to this function:
+//  * There are no active references;
+//  * The mapped bit is in "mapped" state.
+// Otherwise no changes are done to mapped bit and false is returned.
+func (rm *refcountMapped) tryUnmap() bool {
+	if atomic.LoadInt64(&rm.value) != 0 {
+		return false
+	}
+	return atomic.CompareAndSwapInt64(
+		&rm.value,
+		0,
+		1,
+	)
+}