Remove the FIFO prioritizer; use least-loaded over all streams by default

CHANGELOG.md

@@ -8,6 +8,9 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
+- Remove the OTel-Arrow exporter FIFO prioritizer.  Let "leastloaded" imply least-loaded
+  over all streams and use this behavior by default. [#186](https://github.com/open-telemetry/otel-arrow/pull/186)
+
 - Refactor otelarrowreceiver to do stream.Recv, request processing, and stream.Send in separate goroutines. [#181](https://github.com/open-telemetry/otel-arrow/pull/181)
 
 - Add a semaphore package to limit bytes admitted and total number of waiters. [#174](https://github.com/open-telemetry/otel-arrow/pull/174)

collector/exporter/otelarrowexporter/internal/arrow/README.md

@@ -108,12 +108,6 @@ restart following unavaiable, instead the manager waits for downgrade.
 
 ### Prioritizers
 
-#### FIFO
-
-This prioritizer gives work to the first stream that is ready for it.
-The implementation shares one `chan writeItem` between multiple
-`streamWorkState` objects.
-
 #### LeastLoadedN
 
 This prioritizer randomly selects N active streams and chooses the one

collector/exporter/otelarrowexporter/internal/arrow/exporter_test.go

@@ -34,7 +34,7 @@ import (
 	"google.golang.org/grpc/metadata"
 )
 
-var AllPrioritizers = []PrioritizerName{FifoPrioritizer, LeastLoadedTwoPrioritizer}
+var AllPrioritizers = []PrioritizerName{LeastLoadedPrioritizer, LeastLoadedTwoPrioritizer}
 
 const defaultMaxStreamLifetime = 11 * time.Second
 
@@ -786,30 +786,6 @@ func TestArrowExporterStreamLifetimeAndShutdown(t *testing.T) {
 	}
 }
 
-func BenchmarkFifo4(b *testing.B) {
-	benchmarkPrioritizer(b, 4, FifoPrioritizer)
-}
-
-func BenchmarkFifo8(b *testing.B) {
-	benchmarkPrioritizer(b, 8, FifoPrioritizer)
-}
-
-func BenchmarkFifo16(b *testing.B) {
-	benchmarkPrioritizer(b, 16, FifoPrioritizer)
-}
-
-func BenchmarkFifo32(b *testing.B) {
-	benchmarkPrioritizer(b, 32, FifoPrioritizer)
-}
-
-func BenchmarkFifo64(b *testing.B) {
-	benchmarkPrioritizer(b, 64, FifoPrioritizer)
-}
-
-func BenchmarkFifo128(b *testing.B) {
-	benchmarkPrioritizer(b, 128, FifoPrioritizer)
-}
-
 func BenchmarkLeastLoadedTwo4(b *testing.B) {
 	benchmarkPrioritizer(b, 4, LeastLoadedTwoPrioritizer)
 }

collector/exporter/otelarrowexporter/internal/arrow/prioritizer.go

@@ -21,14 +21,13 @@ type PrioritizerName string
 var _ component.ConfigValidator = PrioritizerName("")
 
 const (
-	DefaultPrioritizer         PrioritizerName = FifoPrioritizer
-	FifoPrioritizer            PrioritizerName = "fifo"
+	DefaultPrioritizer         PrioritizerName = LeastLoadedPrioritizer
+	LeastLoadedPrioritizer     PrioritizerName = llPrefix
 	LeastLoadedTwoPrioritizer  PrioritizerName = llPrefix + "2"
 	LeastLoadedFourPrioritizer PrioritizerName = llPrefix + "4"
 	unsetPrioritizer           PrioritizerName = ""
 
-	llPrefix            = "leastloaded"
-	defaultLeastLoadedN = 4
+	llPrefix = "leastloaded"
 )
 
 // streamPrioritizer is an interface for prioritizing multiple
@@ -56,14 +55,13 @@ func newStreamPrioritizer(dc doneCancel, name PrioritizerName, numStreams int) (
 		name = DefaultPrioritizer
 	}
 	if strings.HasPrefix(string(name), llPrefix) {
-		// error was checked and reported in Validate; in this function,
+		// error was checked and reported in Validate
 		n, err := strconv.Atoi(string(name[len(llPrefix):]))
-		if err != nil {
-			n = defaultLeastLoadedN
+		if err == nil {
+			return newBestOfNPrioritizer(dc, n, numStreams, pendingRequests)
 		}
-		return newBestOfNPrioritizer(dc, n, numStreams, pendingRequests)
 	}
-	return newFifoPrioritizer(dc, numStreams)
+	return newBestOfNPrioritizer(dc, numStreams, numStreams, pendingRequests)
 }
 
 // pendingRequests is the load function used by leastloadedN.
@@ -76,9 +74,11 @@ func pendingRequests(sws *streamWorkState) float64 {
 // Validate implements component.ConfigValidator
 func (p PrioritizerName) Validate() error {
 	switch p {
-	case FifoPrioritizer, unsetPrioritizer:
+	// Exact match cases
+	case LeastLoadedPrioritizer, unsetPrioritizer:
 		return nil
 	}
+	// "leastloadedN" cases
 	if !strings.HasPrefix(string(p), llPrefix) {
 		return fmt.Errorf("unrecognized prioritizer: %q", string(p))
 	}