go-04-map-concurrent

# Concurrent ordered map

Implement **`solution.go`** in `package challenge` exporting:

```go
func MapConcurrent(inputs []int, workers int, fn func(int) int) []int
```

Apply `fn` to every element of `inputs` using a pool of **at most `workers`**
goroutines running concurrently, and return the results.

Rules:

- The returned slice has the **same length** as `inputs`, and `result[i]` is
  `fn(inputs[i])` — i.e. results are in the **same order as the inputs**,
  regardless of the order goroutines finish in.
- Work must be distributed across concurrent goroutines (use goroutines plus
  channels and/or `sync.WaitGroup`). No more than `workers` goroutines may be
  processing elements at the same time.
- `workers` may be larger than `len(inputs)`; never start more workers than
  there is work for, and never start fewer than one when there is work.
- If `inputs` is empty (or `nil`), return an empty (non-nil is fine, but it must
  have length 0) slice without starting any work.
- You may assume `workers >= 1` and that `fn` is safe to call concurrently
  (it does not share mutable state).

Examples:

- `MapConcurrent([]int{1, 2, 3}, 2, func(x int) int { return x * x })` → `[]int{1, 4, 9}`
- `MapConcurrent([]int{}, 4, fn)` → `[]int{}`
- `MapConcurrent([]int{5}, 8, func(x int) int { return x + 1 })` → `[]int{6}`

package challenge

import (
	"reflect"
	"sync"
	"sync/atomic"
	"testing"
)

func TestMapConcurrentMatchesSequential(t *testing.T) {
	fn := func(x int) int { return x*x + 1 }
	cases := []struct {
		name    string
		inputs  []int
		workers int
	}{
		{"empty", []int{}, 4},
		{"nil", nil, 4},
		{"single", []int{7}, 1},
		{"single many workers", []int{7}, 16},
		{"workers one", []int{1, 2, 3, 4, 5}, 1},
		{"workers equal len", []int{1, 2, 3, 4}, 4},
		{"workers gt len", []int{1, 2, 3}, 10},
		{"larger", []int{9, 8, 7, 6, 5, 4, 3, 2, 1, 0, -1, -2}, 3},
		{"negatives", []int{-5, -4, -3, -2, -1}, 2},
	}
	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			want := make([]int, len(c.inputs))
			for i, v := range c.inputs {
				want[i] = fn(v)
			}
			got := MapConcurrent(c.inputs, c.workers, fn)
			if len(got) != len(c.inputs) {
				t.Fatalf("len = %d, want %d", len(got), len(c.inputs))
			}
			if !reflect.DeepEqual(got, want) {
				t.Errorf("MapConcurrent(%v, %d) = %v, want %v", c.inputs, c.workers, got, want)
			}
		})
	}
}

func TestMapConcurrentOrderPreserved(t *testing.T) {
	// Identity maps input value to output position; any reordering would be caught.
	inputs := make([]int, 200)
	for i := range inputs {
		inputs[i] = i * 3
	}
	got := MapConcurrent(inputs, 7, func(x int) int { return x })
	for i, v := range inputs {
		if got[i] != v {
			t.Fatalf("index %d = %d, want %d (order not preserved)", i, got[i], v)
		}
	}
}

func TestMapConcurrentEmptyReturnsLenZero(t *testing.T) {
	got := MapConcurrent(nil, 4, func(x int) int { return x })
	if len(got) != 0 {
		t.Fatalf("len = %d, want 0", len(got))
	}
}

func TestMapConcurrentRespectsWorkerLimit(t *testing.T) {
	inputs := make([]int, 100)
	for i := range inputs {
		inputs[i] = i
	}
	const limit = 4

	var mu sync.Mutex
	cond := sync.NewCond(&mu)
	var active int64
	var maxActive int64
	var arrived int

	// fn blocks until either `limit` callers are concurrently in flight or no more
	// callers can arrive. This forces real concurrency while bounding it at `limit`:
	// if the implementation ran fewer than `limit` workers it would deadlock-stall,
	// and any run above `limit` would be recorded in maxActive.
	fn := func(x int) int {
		cur := atomic.AddInt64(&active, 1)
		for {
			m := atomic.LoadInt64(&maxActive)
			if cur <= m || atomic.CompareAndSwapInt64(&maxActive, m, cur) {
				break
			}
		}
		mu.Lock()
		arrived++
		if arrived >= limit {
			cond.Broadcast()
		}
		for arrived < limit {
			cond.Wait()
		}
		mu.Unlock()
		atomic.AddInt64(&active, -1)
		return x * 2
	}

	got := MapConcurrent(inputs, limit, fn)
	if m := atomic.LoadInt64(&maxActive); m > limit {
		t.Fatalf("observed %d concurrent workers, limit was %d", m, limit)
	}
	if atomic.LoadInt64(&maxActive) < 2 {
		t.Fatalf("no real concurrency observed (maxActive=%d)", maxActive)
	}
	for i := range inputs {
		if got[i] != inputs[i]*2 {
			t.Fatalf("index %d = %d, want %d", i, got[i], inputs[i]*2)
		}
	}
}

func TestMapConcurrentCallsFnOncePerElement(t *testing.T) {
	inputs := []int{10, 20, 30, 40, 50, 60}
	var calls int64
	got := MapConcurrent(inputs, 3, func(x int) int {
		atomic.AddInt64(&calls, 1)
		return x + 1
	})
	if calls != int64(len(inputs)) {
		t.Fatalf("fn called %d times, want %d", calls, len(inputs))
	}
	for i, v := range inputs {
		if got[i] != v+1 {
			t.Errorf("index %d = %d, want %d", i, got[i], v+1)
		}
	}
}

You are an expert programmer. Solve the task exactly as specified. Output your solution as fenced code blocks using the required file name(s) and the exact function/type signatures requested. Prefer correctness; do not include prose outside code unless asked.

```go
package challenge

import (
	"sync"
)

func MapConcurrent(inputs []int, workers int, fn func(int) int) []int {
	if len(inputs) == 0 {
		return []int{}
	}

	// Cap workers to the number of inputs
	if workers > len(inputs) {
		workers = len(inputs)
	}

	// Create a channel to send work items
	work := make(chan int, len(inputs))
	
	// Create a channel to receive results
	results := make(chan struct {
		index int
		value int
	}, len(inputs))

	var wg sync.WaitGroup

	// Start workers
	for i := 0; i < workers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for index := range work {
				value := fn(index)
				results <- struct {
					index int
					value int
				}{index: index, value: value}
			}
		}()
	}

	// Send all work items
	for i, input := range inputs {
		work <- i // We send the index to avoid capturing the loop variable
	}
	close(work)

	// Collect results
	finalResults := make([]int, len(inputs))
	
	// Wait for all workers to finish
	go func() {
		wg.Wait()
		close(results)
	}()

	// Collect results in order
	for result := range results {
		finalResults[result.index] = result.value
	}

	return finalResults
}
```

[
  {
    "total": 1,
    "answer": "```go\npackage challenge\n\nimport (\n\t\"sync\"\n)\n\nfunc MapConcurrent(inputs []int, workers int, fn func(int) int) []int {\n\tif len(inputs) == 0 {\n\t\treturn []int{}\n\t}\n\n\t// Cap workers to the number of inputs\n\tif workers > len(inputs) {\n\t\tworkers = len(inputs)\n\t}\n\n\t// Create a channel to send work items\n\twork := make(chan int, len(inputs))\n\t\n\t// Create a channel to receive results\n\tresults := make(chan struct {\n\t\tindex int\n\t\tvalue int\n\t}, len(inputs))\n\n\tvar wg sync.WaitGroup\n\n\t// Start workers\n\tfor i := 0; i < workers; i++ {\n\t\twg.Add(1)\n\t\tgo func() {\n\t\t\tdefer wg.Done()\n\t\t\tfor index := range work {\n\t\t\t\tvalue := fn(index)\n\t\t\t\tresults <- struct {\n\t\t\t\t\tindex int\n\t\t\t\t\tvalue int\n\t\t\t\t}{index: index, value: value}\n\t\t\t}\n\t\t}()\n\t}\n\n\t// Send all work items\n\tfor i, input := range inputs {\n\t\twork <- i // We send the index to avoid capturing the loop variable\n\t}\n\tclose(work)\n\n\t// Collect results\n\tfinalResults := make([]int, len(inputs))\n\t\n\t// Wait for all workers to finish\n\tgo func() {\n\t\twg.Wait()\n\t\tclose(results)\n\t}()\n\n\t// Collect results in order\n\tfor result := range results {\n\t\tfinalResults[result.index] = result.value\n\t}\n\n\treturn finalResults\n}\n```",
    "passed": 0,
    "test_error": "{\"ImportPath\":\"challenge [challenge.test]\",\"Action\":\"build-output\",\"Output\":\"# challenge [challenge.test]\\n\"}\n{\"ImportPath\":\"challenge [challenge.test]\",\"Action\":\"build-output\",\"Output\":\"./solution.go:44:9: declared and not used: input\\n\"}\n{\"ImportPath\":\"challenge [challenge.test]\",\"Action\":\"build-fail\"}\n{\"Time\":\"2026-07-03T00:39:27.991317887+02:00\",\"Action\":\"start\",\"Package\":\"challenge\"}\n{\"Time\":\"2026-07-03T00:39:27.991375627+02:00\",\"Action\":\"output\",\"Package\":\"challenge\",\"Output\":\"FAIL\\tchallenge [build failed]\\n\"}\n{\"Time\":\"2026-07-03T00:39:27.991379607+02:00\",\"Action\":\"fail\",\"Package\":\"challenge\",\"Elapsed\":0,\"FailedBuild\":\"challenge [challenge.test]\"}"
  },
  {
    "total": 5,
    "answer": "Looking at the error, there's a variable `input` that's declared but not used. Let me analyze the code more carefully and fix it.\n\nThe issue is in my implementation - I'm sending indices to the work channel but then trying to use the actual input values in the function. Also, I need to be more careful about how I handle the indexing.\n\nLet me rewrite this properly:\n\n```go\npackage challenge\n\nimport (\n\t\"sync\"\n)\n\nfunc MapConcurrent(inputs []int, workers int, fn func(int) int) []int {\n\tif len(inputs) == 0 {\n\t\treturn []int{}\n\t}\n\n\t// Cap workers to the number of inputs\n\tif workers > len(inputs) {\n\t\tworkers = len(inputs)\n\t}\n\n\t// Create a channel to send work items (indices)\n\twork := make(chan int, len(inputs))\n\t\n\t// Create a channel to receive results\n\tresults := make(chan struct {\n\t\tindex int\n\t\tvalue int\n\t}, len(inputs))\n\n\tvar wg sync.WaitGroup\n\n\t// Start workers\n\tfor i := 0; i < workers; i++ {\n\t\twg.Add(1)\n\t\tgo func() {\n\t\t\tdefer wg.Done()\n\t\t\tfor idx := range work {\n\t\t\t\t// Apply function to the actual input value at this index\n\t\t\t\tvalue := fn(inputs[idx])\n\t\t\t\tresults <- struct {\n\t\t\t\t\tindex int\n\t\t\t\t\tvalue int\n\t\t\t\t}{index: idx, value: value}\n\t\t\t}\n\t\t}()\n\t}\n\n\t// Send all work items (indices)\n\tfor i := range inputs {\n\t\twork <- i\n\t}\n\tclose(work)\n\n\t// Collect results\n\tfinalResults := make([]int, len(inputs))\n\t\n\t// Wait for all workers to finish\n\tgo func() {\n\t\twg.Wait()\n\t\tclose(results)\n\t}()\n\n\t// Collect results in order\n\tfor result := range results {\n\t\tfinalResults[result.index] = result.value\n\t}\n\n\treturn finalResults\n}\n```\n\nWait, I see the issue now. I'm using `inputs[idx]` inside the goroutine but that's not safe since it's accessing a shared variable. Let me fix this properly:\n\n```go\npackage challenge\n\nimport (\n\t\"sync\"\n)\n\nfunc MapConcurrent(inputs []int, workers int, fn func(int) int) []int {\n\tif len(inputs) == 0 {\n\t\treturn []int{}\n\t}\n\n\t// Cap workers to the number of inputs\n\tif workers > len(inputs) {\n\t\tworkers = len(inputs)\n\t}\n\n\t// Create a channel to send work items (indices)\n\twork := make(chan int, len(inputs))\n\t\n\t// Create a channel to receive results\n\tresults := make(chan struct {\n\t\tindex int\n\t\tvalue int\n\t}, len(inputs))\n\n\tvar wg sync.WaitGroup\n\n\t// Start workers\n\tfor i := 0; i < workers; i++ {\n\t\twg.Add(1)\n\t\tgo func() {\n\t\t\tdefer wg.Done()\n\t\t\tfor idx := range work {\n\t\t\t\t// Apply function to the actual input value at this index\n\t\t\t\tvalue := fn(inputs[idx])\n\t\t\t\tresults <- struct {\n\t\t\t\t\tindex int\n\t\t\t\t\tvalue int\n\t\t\t\t}{index: idx, value: value}\n\t\t\t}\n\t\t}()\n\t}\n\n\t// Send all work items (indices)\n\tfor i := range inputs {\n\t\twork <- i\n\t}\n\tclose(work)\n\n\t// Collect results\n\tfinalResults := make([]int, len(inputs))\n\t\n\t// Wait for all workers to finish\n\tgo func() {\n\t\twg.Wait()\n\t\tclose(results)\n\t}()\n\n\t// Collect results in order\n\tfor result := range results {\n\t\tfinalResults[result.index] = result.value\n\t}\n\n\treturn finalResults\n}\n```",
    "passed": 5,
    "test_error": ""
  }
]