ronantakizawa/codereview-bench huggingface dataset:
Repository: containerd/containerd
File: core/mount/mount_linux.go (Go)

Code before:
	if err := unix.Statfs(path, &statfs); err != nil {
		return 0, err
	}

	// The set of keys come from https://github.com/torvalds/linux/blob/v4.13/fs/namespace.c#L1034-L1048.
	unprivilegedFlags := []int{
		unix.MS_RDONLY,
		unix.MS_NODEV,
		unix.MS_NOEXEC,
		unix.MS_NOSUID,
		unix.MS_NOATIME,
		unix.MS_RELATIME,
		unix.MS_NODIRATIME,
	}

	var flags int
	for flag := range unprivilegedFlags {
		if int(statfs.Flags)&flag == flag {
			flags |= flag
		}
	}

	return flags, nil
}

func doPrepareIDMappedOverlay(tmpDir string, lowerDirs []string, usernsFd int) (tmpLowerDirs []string, cleanup func(), retErr error) {
	commonDir, err := getCommonDirectory(lowerDirs)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to determine common parent: %w", err)
	}

	tempRemountsLocation, err := os.MkdirTemp(tmpDir, "ovl-idmapped")
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create temporary overlay lowerdir mount location: %w", err)
	}
	cleanDir := func() {
		if err := os.Remove(tempRemountsLocation); err != nil {
			log.L.WithError(err).Infof("failed to remove idmapped directory")
		}
	}
	defer func() {
		if retErr != nil {
			cleanDir()
		}
	}()

	// IDMapMount the directory containing all the layers
	if err := IDMapMountWithAttrs(commonDir, tempRemountsLocation, usernsFd, unix.MOUNT_ATTR_RDONLY, 0); err != nil {
		return nil, nil, err
	}
	cleanMount := func() {

Review comment (suggestion):
we can assign a name to the returned error, such as retErr. But for `tmpLowerDirs` and `cleanup`, we should use `_`, since it’s only needed at the end of the function. Otherwise, keeping the variable might cause confusion. For example, `cleanup` and `cleanDir`.

Code after:
	if err := unix.Statfs(path, &statfs); err != nil {
		return 0, err
	}

	// The set of keys come from https://github.com/torvalds/linux/blob/v4.13/fs/namespace.c#L1034-L1048.
	unprivilegedFlags := []int{
		unix.MS_RDONLY,
		unix.MS_NODEV,
		unix.MS_NOEXEC,
		unix.MS_NOSUID,
		unix.MS_NOATIME,
		unix.MS_RELATIME,
		unix.MS_NODIRATIME,
	}

	var flags int
	for flag := range unprivilegedFlags {
		if int(statfs.Flags)&flag == flag {
			flags |= flag
		}
	}

	return flags, nil
}

func doPrepareIDMappedOverlay(tmpDir string, lowerDirs []string, usernsFd int) (_ []string, _ func(), retErr error) {
	commonDir, err := getCommonDirectory(lowerDirs)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to determine common parent: %w", err)
	}

	tempRemountsLocation, err := os.MkdirTemp(tmpDir, "ovl-idmapped")
	if err != nil {
		return nil, nil, fmt.Errorf("failed to create temporary overlay lowerdir mount location: %w", err)
	}
	cleanDir := func() {
		if err := os.Remove(tempRemountsLocation); err != nil {
			log.L.WithError(err).Infof("failed to remove idmapped directory")
		}
	}
	defer func() {
		if retErr != nil {
			cleanDir()
		}
	}()

	// IDMapMount the directory containing all the layers
	if err := IDMapMountWithAttrs(commonDir, tempRemountsLocation, usernsFd, unix.MOUNT_ATTR_RDONLY, 0); err != nil {
		return nil, nil, err
	}
	cleanMount := func() {
Repository: containerd/containerd
File: core/runtime/v2/shim_load.go (Go)

Code before:
}

func (m *ShimManager) loadShims(ctx context.Context, stateDir string) error {
	ns, err := namespaces.NamespaceRequired(ctx)
	if err != nil {
		return err
	}
	ctx = log.WithLogger(ctx, log.G(ctx).WithField("namespace", ns))

	shimDirs, err := os.ReadDir(filepath.Join(stateDir, ns))
	if err != nil {
		return err
	}
	eg, ctx2 := errgroup.WithContext(ctx)
	eg.SetLimit(100)
	for _, sd := range shimDirs {
		eg.Go(func() error {
			if !sd.IsDir() {
				return nil
			}
			id := sd.Name()
			// skip hidden directories
			if len(id) > 0 && id[0] == '.' {
				return nil
			}
			bundle, err := LoadBundle(ctx2, stateDir, id)
			if err != nil {
				// fine to return error here, it is a programmer error if the context
				// does not have a namespace
				return err
			}
			// fast path
			f, err := os.Open(bundle.Path)
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}

			bf, err := f.Readdirnames(-1)
			f.Close()
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}
			if len(bf) == 0 {
				bundle.Delete()
				return nil
			}
			if err := m.loadShim(ctx2, bundle); err != nil {

Review comment (suggestion):
Just want to mention one thing: except this error, we should not return any error in eg.Go. If one background goroutine returns error, eg.Go will cancel that context which impacts other running goroutines. It could result in unexpected issue: cleaning-up recoverable and running containers. 

Code after:
	return nil
}

func (m *ShimManager) loadShims(ctx context.Context, stateDir string) error {
	ns, err := namespaces.NamespaceRequired(ctx)
	if err != nil {
		return err
	}
	ctx = log.WithLogger(ctx, log.G(ctx).WithField("namespace", ns))

	shimDirs, err := os.ReadDir(filepath.Join(stateDir, ns))
	if err != nil {
		return err
	}
	eg, ctx2 := errgroup.WithContext(ctx)
	eg.SetLimit(runtime.GOMAXPROCS(0))
	var errLoad error
	for _, sd := range shimDirs {
		if !sd.IsDir() {
			continue
		}
		id := sd.Name()
		// skip hidden directories
		if len(id) > 0 && id[0] == '.' {
			continue
		}
		bundle, err := LoadBundle(ctx, stateDir, id)
		if err != nil {
			errLoad = err
			// fine to return error here, it is a programmer error if the context
			// does not have a namespace
			break
		}
		eg.Go(func() error {
			// fast path
			f, err := os.Open(bundle.Path)
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}

			bf, err := f.Readdirnames(-1)
			f.Close()
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}
			if len(bf) == 0 {
				bundle.Delete()
Repository: containerd/containerd
File: core/runtime/v2/shim_load.go (Go)

Code before:
	return nil
}

func (m *ShimManager) loadShims(ctx context.Context, stateDir string) error {
	ns, err := namespaces.NamespaceRequired(ctx)
	if err != nil {
		return err
	}
	ctx = log.WithLogger(ctx, log.G(ctx).WithField("namespace", ns))

	shimDirs, err := os.ReadDir(filepath.Join(stateDir, ns))
	if err != nil {
		return err
	}
	eg, ctx2 := errgroup.WithContext(ctx)
	eg.SetLimit(runtime.GOMAXPROCS(0))
	for _, sd := range shimDirs {
		eg.Go(func() error {
			if !sd.IsDir() {
				return nil
			}
			id := sd.Name()
			// skip hidden directories
			if len(id) > 0 && id[0] == '.' {
				return nil
			}
			bundle, err := LoadBundle(ctx2, stateDir, id)
			if err != nil {
				// fine to return error here, it is a programmer error if the context
				// does not have a namespace
				return err
			}
			// fast path
			f, err := os.Open(bundle.Path)
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}

			bf, err := f.Readdirnames(-1)
			f.Close()
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}
			if len(bf) == 0 {
				bundle.Delete()
				return nil
			}

Review comment (refactor):
nit: can we move this out of goroutine and just continue to the next for loop? I feel it's not worthwhile to create goroutines for those cases.

Code after:
	return nil
}

func (m *ShimManager) loadShims(ctx context.Context, stateDir string) error {
	ns, err := namespaces.NamespaceRequired(ctx)
	if err != nil {
		return err
	}
	ctx = log.WithLogger(ctx, log.G(ctx).WithField("namespace", ns))

	shimDirs, err := os.ReadDir(filepath.Join(stateDir, ns))
	if err != nil {
		return err
	}
	eg, ctx2 := errgroup.WithContext(ctx)
	eg.SetLimit(runtime.GOMAXPROCS(0))
	var errLoad error
	for _, sd := range shimDirs {
		if !sd.IsDir() {
			continue
		}
		id := sd.Name()
		// skip hidden directories
		if len(id) > 0 && id[0] == '.' {
			continue
		}
		bundle, err := LoadBundle(ctx, stateDir, id)
		if err != nil {
			errLoad = err
			// fine to return error here, it is a programmer error if the context
			// does not have a namespace
			break
		}
		eg.Go(func() error {
			// fast path
			f, err := os.Open(bundle.Path)
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}

			bf, err := f.Readdirnames(-1)
			f.Close()
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}
			if len(bf) == 0 {
				bundle.Delete()
Repository: containerd/containerd
File: core/runtime/v2/shim_load.go (Go)

Code before:
	}
	ctx = log.WithLogger(ctx, log.G(ctx).WithField("namespace", ns))

	shimDirs, err := os.ReadDir(filepath.Join(stateDir, ns))
	if err != nil {
		return err
	}
	eg, ctx2 := errgroup.WithContext(ctx)
	eg.SetLimit(runtime.GOMAXPROCS(0))
	var lock sync.Mutex
	var errLoad error
	for _, sd := range shimDirs {
		if !sd.IsDir() {
			continue
		}
		id := sd.Name()
		// skip hidden directories
		if len(id) > 0 && id[0] == '.' {
			continue
		}
		eg.Go(func() error {
			bundle, err := LoadBundle(ctx2, stateDir, id)
			if err != nil {
				// fine to collect error here, it is a programmer error if the context
				// does not have a namespace
				lock.Lock()
				defer lock.Unlock()
				errors.Join(errLoad, err)
				// don't interrupt other goroutines' execution
				return nil
			}
			// fast path
			f, err := os.Open(bundle.Path)
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}

			bf, err := f.Readdirnames(-1)
			f.Close()
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}
			if len(bf) == 0 {
				bundle.Delete()
				return nil
			}
			if err := m.loadShim(ctx2, bundle); err != nil {

Review comment (suggestion):
Suggestion: remove that `lock` 

```go
var errLoad error
for _, sd := range shimDirs {
        ...
        bundle, err := LoadBundle(ctx, stateDir, id)
        if err != nil {
                errLoad = err
                break
        }

        eg.Go(func() error {
                ...
        })
         ...
}
 _ = eg.Wait()
return errLoad
```

Code after:
	ctx = log.WithLogger(ctx, log.G(ctx).WithField("namespace", ns))

	shimDirs, err := os.ReadDir(filepath.Join(stateDir, ns))
	if err != nil {
		return err
	}
	eg, ctx2 := errgroup.WithContext(ctx)
	eg.SetLimit(runtime.GOMAXPROCS(0))
	var errLoad error
	for _, sd := range shimDirs {
		if !sd.IsDir() {
			continue
		}
		id := sd.Name()
		// skip hidden directories
		if len(id) > 0 && id[0] == '.' {
			continue
		}
		bundle, err := LoadBundle(ctx, stateDir, id)
		if err != nil {
			errLoad = err
			// fine to return error here, it is a programmer error if the context
			// does not have a namespace
			break
		}
		eg.Go(func() error {
			// fast path
			f, err := os.Open(bundle.Path)
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}

			bf, err := f.Readdirnames(-1)
			f.Close()
			if err != nil {
				bundle.Delete()
				log.G(ctx2).WithError(err).Errorf("fast path read bundle path for %s", bundle.Path)
				return nil
			}
			if len(bf) == 0 {
				bundle.Delete()
				return nil
			}
			if err := m.loadShim(ctx2, bundle); err != nil {
				log.G(ctx2).WithError(err).Errorf("failed to load shim %s", bundle.Path)
				bundle.Delete()
				return nil
			}
			return nil
Repository: containerd/containerd
File: integration/restart_linux_test.go (Go)

Code before:
	require.NoError(t, ctrd.isReady())

	podCtxs := []*podTCtx{}
	defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
	}()

	for i := 0; i < 100; i++ {
		podCtx := newPodTCtx(t,
			ctrd.criRuntimeService(t),
			fmt.Sprintf("test-restart-%d", i),
			"sandbox",
			WithHostNetwork)

		podCtxs = append(podCtxs, podCtx)
	}

	assert.NoError(t, ctrd.kill(syscall.SIGTERM))
	assert.NoError(t, ctrd.wait(5*time.Minute))

	ctrd = newCtrdProc(t, *containerdBin, workDir, []string{})
	require.NoError(t, ctrd.isReady())

	t.Cleanup(func() {
		assert.NoError(t, ctrd.kill(syscall.SIGTERM))
		assert.NoError(t, ctrd.wait(5*time.Minute))
	})
}


Review comment (suggestion):
please remove that t.Cleanup and call 

```
assert.NoError(t, ctrd.kill(syscall.SIGTERM))
		assert.NoError(t, ctrd.wait(5*time.Minute))
```

Code after:
	require.NoError(t, ctrd.isReady())

	podCtxs := []*podTCtx{}
	defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
		assert.NoError(t, ctrd.kill(syscall.SIGTERM))
		assert.NoError(t, ctrd.wait(5*time.Minute))
	}()

	for i := 0; i < 100; i++ {
		podCtx := newPodTCtx(t,
			ctrd.criRuntimeService(t),
			fmt.Sprintf("test-restart-%d", i),
			"sandbox",
			WithHostNetwork)

		podCtxs = append(podCtxs, podCtx)
	}

	assert.NoError(t, ctrd.kill(syscall.SIGTERM))
	assert.NoError(t, ctrd.wait(5*time.Minute))

	ctrd = newCtrdProc(t, *containerdBin, workDir, []string{})
	require.NoError(t, ctrd.isReady())
}
Repository: containerd/containerd
File: integration/restart_linux_test.go (Go)

Code before:
			assert.NoError(t, err)
		}
	}
	assert.True(t, foundUnkownSb)
}

func TestRestartContainerdTime(t *testing.T) {
	workDir := t.TempDir()

	currentReleaseCtrdDefaultConfig(t, workDir)

	ctrd := newCtrdProc(t, *containerdBin, workDir, []string{})

	logPath := ctrd.logPath()
	t.Cleanup(func() {
		if t.Failed() {
			dumpFileContent(t, logPath)
		}
	})
	require.NoError(t, ctrd.isReady())

	podCtxs := []*podTCtx{}
	defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
	}()

	for i := 0; i < 100; i++ {
		podCtx := newPodTCtx(t,
			ctrd.criRuntimeService(t),
			fmt.Sprintf("test-restart-%d", i),
			"sandbox",
			WithHostNetwork)

		podCtxs = append(podCtxs, podCtx)
	}

	assert.NoError(t, ctrd.kill(syscall.SIGTERM))
	assert.NoError(t, ctrd.wait(5*time.Minute))

	ctrd = newCtrdProc(t, *containerdBin, workDir, []string{})
	require.NoError(t, ctrd.isReady())
}


Review comment (suggestion):
sorry for that my last comment. we do need the cleanup.

Maybe we should call it here.

```
defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
         assert.NoError(t, ctrd.kill(syscall.SIGTERM))
	assert.NoError(t, ctrd.wait(5*time.Minute))
}()
```


Code after:
			assert.NoError(t, err)
		}
	}
	assert.True(t, foundUnkownSb)
}

func TestReload100Pods(t *testing.T) {
	workDir := t.TempDir()

	currentReleaseCtrdDefaultConfig(t, workDir)

	ctrd := newCtrdProc(t, *containerdBin, workDir, []string{})

	logPath := ctrd.logPath()
	t.Cleanup(func() {
		if t.Failed() {
			dumpFileContent(t, logPath)
		}
	})
	require.NoError(t, ctrd.isReady())

	podCtxs := []*podTCtx{}
	defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
		assert.NoError(t, ctrd.kill(syscall.SIGTERM))
		assert.NoError(t, ctrd.wait(5*time.Minute))
	}()

	for i := 0; i < 100; i++ {
		podCtx := newPodTCtx(t,
			ctrd.criRuntimeService(t),
			fmt.Sprintf("test-restart-%d", i),
			"sandbox",
			WithHostNetwork)

		podCtxs = append(podCtxs, podCtx)
	}

	assert.NoError(t, ctrd.kill(syscall.SIGTERM))
	assert.NoError(t, ctrd.wait(5*time.Minute))

	ctrd = newCtrdProc(t, *containerdBin, workDir, []string{})
	require.NoError(t, ctrd.isReady())
}
Repository: containerd/containerd
File: integration/restart_linux_test.go (Go)

Code before:
		RestartContainerd(t, syscall.SIGTERM)
		waitCh <- struct{}{}
	}()
	t.Logf("Create a sandbox with shim create delay")
	_, err = runtimeService.RunPodSandbox(sbUnknownConfig, failpointRuntimeHandler)
	assert.Error(t, err)
	<-waitCh
	sbs, err := runtimeService.ListPodSandbox(nil)
	assert.NoError(t, err)
	foundUnkownSb := false
	for _, sb := range sbs {
		if sb.Metadata.Name == "sandbox_unknown" {
			foundUnkownSb = true
		}
		if status, ok := sbStatuses[sb.Metadata.Name]; ok {
			assert.Equal(t, status, sb.State)
			err = runtimeService.StopPodSandbox(sb.Id)
			assert.NoError(t, err)
			err = runtimeService.RemovePodSandbox(sb.Id)
			assert.NoError(t, err)
		}
	}
	assert.True(t, foundUnkownSb)
}

func TestRestartContainerdTime(t *testing.T) {
	workDir := t.TempDir()

	currentReleaseCtrdDefaultConfig(t, workDir)

	ctrd := newCtrdProc(t, *containerdBin, workDir, []string{})

	logPath := ctrd.logPath()
	t.Cleanup(func() {
		if t.Failed() {
			dumpFileContent(t, logPath)
		}
	})
	require.NoError(t, ctrd.isReady())

	podCtxs := []*podTCtx{}
	defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
	}()

	for i := 0; i < 100; i++ {
		podCtx := newPodTCtx(t,
			ctrd.criRuntimeService(t),
			fmt.Sprintf("test-restart-%d", i),

Review comment (suggestion):
```suggestion
func TestReload100Pods(t *testing.T) {
```

Code after:
		RestartContainerd(t, syscall.SIGTERM)
		waitCh <- struct{}{}
	}()
	t.Logf("Create a sandbox with shim create delay")
	_, err = runtimeService.RunPodSandbox(sbUnknownConfig, failpointRuntimeHandler)
	assert.Error(t, err)
	<-waitCh
	sbs, err := runtimeService.ListPodSandbox(nil)
	assert.NoError(t, err)
	foundUnkownSb := false
	for _, sb := range sbs {
		if sb.Metadata.Name == "sandbox_unknown" {
			foundUnkownSb = true
		}
		if status, ok := sbStatuses[sb.Metadata.Name]; ok {
			assert.Equal(t, status, sb.State)
			err = runtimeService.StopPodSandbox(sb.Id)
			assert.NoError(t, err)
			err = runtimeService.RemovePodSandbox(sb.Id)
			assert.NoError(t, err)
		}
	}
	assert.True(t, foundUnkownSb)
}

func TestReload100Pods(t *testing.T) {
	workDir := t.TempDir()

	currentReleaseCtrdDefaultConfig(t, workDir)

	ctrd := newCtrdProc(t, *containerdBin, workDir, []string{})

	logPath := ctrd.logPath()
	t.Cleanup(func() {
		if t.Failed() {
			dumpFileContent(t, logPath)
		}
	})
	require.NoError(t, ctrd.isReady())

	podCtxs := []*podTCtx{}
	defer func() {
		for _, p := range podCtxs {
			p.stop(true)
		}
		assert.NoError(t, ctrd.kill(syscall.SIGTERM))
		assert.NoError(t, ctrd.wait(5*time.Minute))
	}()

	for i := 0; i < 100; i++ {
		podCtx := newPodTCtx(t,
Repository: containerd/containerd
File: core/remotes/docker/fetcher.go (Go)

Code before:
		for i := range numChunks {
			readers[i], writers[i] = newPipeWriter(bufPool)
		}
		// keep reference of the initial body value to ensure it is closed
		ibody := body
		go func() {
			defer close(queue)
			for i := range numChunks {
				select {
				case queue <- i:
				case <-done:
					if i == 0 {
						ibody.Close()
					}
					return
				}
			}
		}()
		for range parallelism {
			go func() {
				for i := range queue { // first in first out
					copy := func() error {
						var body io.ReadCloser
						if i == 0 {
							body = ibody
						} else {
							if err := r.Acquire(ctx, 1); err != nil {
								_ = writers[i].CloseWithError(err)
								cancelCtx()
								return err
							}
							defer r.Release(1)
							reqClone := req.clone()
							reqClone.setOffset(offset + i*chunkSize)
							nresp, err := reqClone.doWithRetries(ctx, lastHost, withErrorCheck)
							if err != nil {
								_ = writers[i].CloseWithError(err)
								cancelCtx()
								return err
							}
							body = nresp.Body
						}
						_, err := io.Copy(writers[i], io.LimitReader(body, chunkSize))
						_ = body.Close()
						_ = writers[i].CloseWithError(err)
						if err != nil && err != io.EOF {
							cancelCtx()
							return err
						}
						return nil
					}

Review comment (suggestion):
In order to avoid leaky goroutine, my suggestion is to use errgroup and drain that goroutines in beforeClose.

```
BeforeClose: func() {
				cancelCtx()
			},
```

Code after:
		queue := make(chan int64, parallelism)
		ctx, cancel := context.WithCancel(ctx)
		eg, ctx := errgroup.WithContext(ctx)
		readers, writers := make([]io.Reader, numChunks), make([]*pipeWriter, numChunks)
		bufPool := newbufferPool(chunkSize)
		for i := range numChunks {
			readers[i], writers[i] = newPipeWriter(bufPool)
		}
		// keep reference of the initial body value to ensure it is closed
		ibody := body
		eg.Go(func() error {
			defer close(queue)
			for i := range numChunks {
				select {
				case queue <- i:
				case <-ctx.Done():
					if i == 0 {
						ibody.Close()
					}
					return ctx.Err()
				}
			}
			return nil
		})

		for range parallelism {
			eg.Go(func() error {
				for i := range queue { // first in first out
					copy := func() error {
						var body io.ReadCloser
						if i == 0 {
							body = ibody
						} else {
							if err := r.Acquire(ctx, 1); err != nil {
								_ = writers[i].CloseWithError(err)
								return err
							}
							defer r.Release(1)
							reqClone := req.clone()
							reqClone.setOffset(offset + i*chunkSize)
							nresp, err := reqClone.doWithRetries(ctx, lastHost, withErrorCheck)
							if err != nil {
								_ = writers[i].CloseWithError(err)
								return err
							}
							body = nresp.Body
						}
						_, err := io.Copy(writers[i], io.LimitReader(body, chunkSize))
						_ = body.Close()
						_ = writers[i].CloseWithError(err)
						if err != nil && err != io.EOF {
Repository: containerd/containerd
File: core/remotes/docker/fetcher.go (Go)

Code before:
			parallelism = numChunks
		}
		queue := make(chan int64, parallelism)
		ctx, cancelCtx := context.WithCancel(ctx)
		done := ctx.Done()
		readers, writers := make([]io.Reader, numChunks), make([]*pipeWriter, numChunks)
		bufPool := newbufferPool(chunkSize)
		for i := range numChunks {
			readers[i], writers[i] = newPipeWriter(bufPool)
		}
		// keep reference of the initial body value to ensure it is closed
		ibody := body
		go func() {
			defer close(queue)
			for i := range numChunks {
				select {
				case queue <- i:
				case <-done:
					if i == 0 {
						ibody.Close()
					}
					return
				}
			}
		}()
		eg := errgroup.Group{}
		for range parallelism {
			eg.Go(func() error {
				for i := range queue { // first in first out
					copy := func() error {
						var body io.ReadCloser
						if i == 0 {
							body = ibody
						} else {
							if err := r.Acquire(ctx, 1); err != nil {
								_ = writers[i].CloseWithError(err)
								cancelCtx()
								return err
							}
							defer r.Release(1)
							reqClone := req.clone()
							reqClone.setOffset(offset + i*chunkSize)
							nresp, err := reqClone.doWithRetries(ctx, lastHost, withErrorCheck)
							if err != nil {
								_ = writers[i].CloseWithError(err)
								cancelCtx()
								return err
							}
							body = nresp.Body
						}
						_, err := io.Copy(writers[i], io.LimitReader(body, chunkSize))

Review comment (suggestion):
this should be `g, ctx := errgroup.WithContext(ctx)`

Code after:
		}
		if numChunks < parallelism {
			parallelism = numChunks
		}

		// Prepare channels, buffer pool, and readers/writers for parallel fetching.
		queue := make(chan int64, parallelism)
		ctx, cancel := context.WithCancel(ctx)
		eg, ctx := errgroup.WithContext(ctx)
		readers, writers := make([]io.Reader, numChunks), make([]*pipeWriter, numChunks)
		bufPool := newbufferPool(chunkSize)
		for i := range numChunks {
			readers[i], writers[i] = newPipeWriter(bufPool)
		}
		// keep reference of the initial body value to ensure it is closed
		ibody := body
		eg.Go(func() error {
			defer close(queue)
			for i := range numChunks {
				select {
				case queue <- i:
				case <-ctx.Done():
					if i == 0 {
						ibody.Close()
					}
					return ctx.Err()
				}
			}
			return nil
		})

		for range parallelism {
			eg.Go(func() error {
				for i := range queue { // first in first out
					copy := func() error {
						var body io.ReadCloser
						if i == 0 {
							body = ibody
						} else {
							if err := r.Acquire(ctx, 1); err != nil {
								_ = writers[i].CloseWithError(err)
								return err
							}
							defer r.Release(1)
							reqClone := req.clone()
							reqClone.setOffset(offset + i*chunkSize)
							nresp, err := reqClone.doWithRetries(ctx, lastHost, withErrorCheck)
							if err != nil {
								_ = writers[i].CloseWithError(err)
								return err
							}