refactor: Replace ctx.Done() with ctx.Err() (#19546)

* refactor: Replace ctx.Done() with ctx.Err()

Prior to this commit we checked for context cancellation with a select
block and context.Context.Done() without multiplexing over any other
channel like:

  select {
    case <-ctx.Done():
      // handle cancellation
    default:
      // fallthrough
  }

This commit replaces those type of blocks with a simple check of
ctx.Err().  This has the following benefits:

* Calling ctx.Err() is much faster than entering a select block.

* ctx.Done() allocates a channel when called for the first time.

* Testing the result of ctx.Err() is a reliable way of determininging if
  a context.Context value has been canceled.

* fix: Fix data race in execDeleteTagValueEntry()

influxql/query/executor.go

@@ -311,15 +311,7 @@ LOOP:
 		e.Metrics.Requests.WithLabelValues(statusLabel).Inc()
 
 		// Check if the query was interrupted during an uninterruptible statement.
-		interrupted := false
-		select {
-		case <-ctx.Done():
-			interrupted = true
-		default:
-			// Query has not been interrupted.
-		}
-
-		if interrupted {
+		if err := ctx.Err(); err != nil {
 			statusLabel = control.LabelInterruptedErr
 			e.Metrics.Requests.WithLabelValues(statusLabel).Inc()
 			break

kv/store.go

@@ -249,10 +249,8 @@ func WalkCursor(ctx context.Context, cursor ForwardCursor, visit VisitFunc) (err
 			return err
 		}
 
-		select {
-		case <-ctx.Done():
-			return ctx.Err()
-		default:
+		if err := ctx.Err(); err != nil {
+			return err
 		}
 	}
 

telemetry/push.go

@@ -78,10 +78,10 @@ func (p *Pusher) push(ctx context.Context) error {
 	req.Header.Set("Content-Type", string(p.PushFormat))
 
 	res, err := p.Client.Do(req)
-	select {
-	case <-ctx.Done():
-		return ctx.Err()
-	default:
+
+	// FIXME: consider why we're checking for cancellation here.
+	if err := ctx.Err(); err != nil {
+		return err
 	}
 
 	if err != nil {

tsdb/index/tsi1/log_file.go

@@ -108,6 +108,8 @@ func (f *LogFile) bytes() int {
 
 // Open reads the log from a file and validates all the checksums.
 func (f *LogFile) Open() error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
 	if err := f.open(); err != nil {
 		f.Close()
 		return err
@@ -717,6 +719,7 @@ func (f *LogFile) execSeriesEntry(e *LogEntry) {
 		}
 
 		ts.tagValues[string(v)] = tv
+
 		mm.tagSet[string(k)] = ts
 	}
 

v1/coordinator/statement_executor.go

@@ -203,11 +203,8 @@ func (e *StatementExecutor) executeExplainAnalyzeStatement(ctx context.Context,
 			goto CLEANUP
 		} else if row == nil {
 			// Check if the query was interrupted while emitting.
-			select {
-			case <-ctx.Done():
-				err = ctx.Err()
+			if err = ctx.Err(); err != nil {
 				goto CLEANUP
-			default:
 			}
 			break
 		}
@@ -266,10 +263,8 @@ func (e *StatementExecutor) executeSelectStatement(ctx context.Context, stmt *in
 			return err
 		} else if row == nil {
 			// Check if the query was interrupted while emitting.
-			select {
-			case <-ctx.Done():
-				return ctx.Err()
-			default:
+			if err := ctx.Err(); err != nil {
+				return err
 			}
 			break
 		}