Added counters.Reader.ScanUnsafe to avoid allocations

Calls to counters.Reader.Scan were allocating heap that would be
released after each call. Reader.labelValue's call to
atomic.Buffer.GetBytesArray is make'ing a []byte and then labelValue
would construct a string from that allocation again.

Added ScanUnsafe for a code path that doesn't allocate.  It shares scan
with Scan and scan reuses a []byte on each call.  Scan copies it to a
string which allocates, but remains worry free. ScanUnsafe casts the
slice to a string that is not safe to retain outside the callback.

aeron/counters/reader.go

@@ -123,25 +123,43 @@ func counterOffset(counterId int32) int32 {
 }
 
 func (reader *Reader) Scan(cb func(Counter)) {
+	bytesToString := func(labelBytes []byte) string {
+		return string(labelBytes)
+	}
+	reader.scan(cb, bytesToString)
+}
+
+// ScanUnsafe callbacks must only consume the label argument within  the callback or must deep copy it
+// because the underlying memory be reused for the next counter within scan.
+func (reader *Reader) ScanUnsafe(cb func(Counter)) {
+	bytesToString := func(labelBytes []byte) string {
+		return unsafe.String(&labelBytes[0], len(labelBytes))
+	}
+	reader.scan(cb, bytesToString)
+}
+
+func (reader *Reader) scan(cb func(Counter), bytesToString func(labelBytes []byte) string) {
 
 	var id int32 = 0
 	var i int32 = 0
 
+	label := make([]byte, 0, FullLabelLength)
 	for capacity := reader.metaData.Capacity(); i < capacity; i += MetadataLength {
 		recordStatus := reader.metaData.GetInt32Volatile(i)
 		if RecordUnused == recordStatus {
 			break
 		} else if RecordAllocated == recordStatus {
 			typeId := reader.metaData.GetInt32(i + TypeIdOffset)
-			label := reader.labelValue(i)
+			label = label[:0]
+			label = reader.labelValueUnsafe(i, label)
 
 			// TODO Get the key buffer
 
 			value := reader.values.GetInt64Volatile(counterOffset(id))
 
 			// fmt.Printf("Reading at offset %d; counterState=%d; typeId=%d\n", i, recordStatus, typeId)
 
-			cb(Counter{id, typeId, value, label})
+			cb(Counter{id, typeId, value, bytesToString(label)})
 		}
 		id++
 	}
@@ -285,3 +303,10 @@ func (reader *Reader) labelValue(metaDataOffset int32) string {
 	labelSize := reader.metaData.GetInt32(metaDataOffset + LabelOffset)
 	return string(reader.metaData.GetBytesArray(metaDataOffset+LabelOffset+4, labelSize))
 }
+
+func (reader *Reader) labelValueUnsafe(metaDataOffset int32, label []byte) []byte {
+	labelSize := reader.metaData.GetInt32(metaDataOffset + LabelOffset)
+	label = label[:labelSize]
+	reader.metaData.GetBytes(metaDataOffset+LabelOffset+4, label)
+	return label
+}