From 491d26bd99421f2bb2c1fd21bd49234ea314cef8 Mon Sep 17 00:00:00 2001 From: taewankim <38234018+callaway12@users.noreply.github.com> Date: Wed, 17 Jun 2026 16:29:46 +0900 Subject: [PATCH 1/2] volumes: replace global lock with per-volume locking for concurrent mounts The EFS volume plugin serializes all volume mount/unmount operations behind a single sync.RWMutex. When multiple ECS tasks start concurrently, each requiring EFS volume mounts (15-72s each), queued requests exceed Docker's plugin timeout (~60s), causing CannotCreateContainerError/TaskFailedToStart. This change introduces per-volume mutexes so that mount/unmount I/O for different volumes can proceed in parallel: - AmazonECSVolumePlugin: replace single `lock` with `mapLock` (short-lived, protects map access only) + `volLocks` (per-volume mutexes held during I/O) - ECSVolumeDriver: release driver lock before mount/unmount syscalls since the plugin layer already holds the per-volume lock - Add TestConcurrentMountsDifferentVolumes to verify parallel execution Operations on the same volume remain serialized for correctness. Metadata-only operations (Create, List, Get, Path) continue using the global map lock since they don't perform I/O. Fixes aws/containers-roadmap#2319 --- ecs-init/volumes/ecs_volume_driver.go | 39 ++++++-- ecs-init/volumes/ecs_volume_plugin.go | 100 +++++++++++++-------- ecs-init/volumes/ecs_volume_plugin_test.go | 97 +++++++++++++++++++- 3 files changed, 191 insertions(+), 45 deletions(-) diff --git a/ecs-init/volumes/ecs_volume_driver.go b/ecs-init/volumes/ecs_volume_driver.go index 4828215134f..12cf4baa9cd 100644 --- a/ecs-init/volumes/ecs_volume_driver.go +++ b/ecs-init/volumes/ecs_volume_driver.go @@ -61,11 +61,10 @@ func (e *ECSVolumeDriver) Setup(name string, v *types.Volume) { e.volumeMounts[name] = mnt } -// Create implements ECSVolumeDriver's Create volume method +// Create implements ECSVolumeDriver's Create volume method. +// The lock is held only for map access and validation; the actual mount I/O +// proceeds without the driver lock since the caller holds a per-volume lock. func (e *ECSVolumeDriver) Create(r *driver.CreateRequest) error { - e.lock.Lock() - defer e.lock.Unlock() - mnt := setOptions(r.Options) mnt.Target = r.Path @@ -74,12 +73,26 @@ func (e *ECSVolumeDriver) Create(r *driver.CreateRequest) error { return err } + // Check for duplicates under read lock + e.lock.RLock() + if _, exists := e.volumeMounts[r.Name]; exists { + e.lock.RUnlock() + return fmt.Errorf("volume %s already mounted", r.Name) + } + e.lock.RUnlock() + + // Perform the mount I/O without holding the driver lock seelog.Infof("Mounting volume %s of type %s at path %s", r.Name, mnt.MountType, mnt.Target) err := mnt.Mount() if err != nil { return fmt.Errorf("mounting volume failed: %v", err) } + + // Register the mount under write lock + e.lock.Lock() e.volumeMounts[r.Name] = mnt + e.lock.Unlock() + return nil } @@ -98,27 +111,37 @@ func setOptions(options map[string]string) *MountHelper { return mnt } -// Remove implements ECSVolumeDriver's Remove volume method +// Remove implements ECSVolumeDriver's Remove volume method. +// The unmount I/O proceeds without the driver lock since the caller holds a per-volume lock. func (e *ECSVolumeDriver) Remove(req *driver.RemoveRequest) error { - e.lock.Lock() - defer e.lock.Unlock() + e.lock.RLock() mnt, ok := e.volumeMounts[req.Name] + e.lock.RUnlock() + if !ok { return fmt.Errorf("volume not found") } + + // Perform the unmount I/O without holding the driver lock err := mnt.Unmount() if err != nil { if strings.Contains(err.Error(), notMountedErrMsg) || strings.Contains(err.Error(), noMountPointSpecifiedErrMsg) { seelog.Infof("Unmounting volume %s failed because it's not mounted.", req.Name) + e.lock.Lock() delete(e.volumeMounts, req.Name) + e.lock.Unlock() return nil } return fmt.Errorf("unmounting volume failed: %v", err) } + + e.lock.Lock() delete(e.volumeMounts, req.Name) + e.lock.Unlock() + seelog.Infof("Unmounted volume %s successfully.", req.Name) - return err + return nil } // Method to check if a volume is currently mounted. diff --git a/ecs-init/volumes/ecs_volume_plugin.go b/ecs-init/volumes/ecs_volume_plugin.go index 33dbab80693..47c9fe71a8a 100644 --- a/ecs-init/volumes/ecs_volume_plugin.go +++ b/ecs-init/volumes/ecs_volume_plugin.go @@ -38,7 +38,12 @@ type AmazonECSVolumePlugin struct { volumeDrivers map[string]driver.VolumeDriver volumes map[string]*types.Volume state *StateManager - lock sync.RWMutex + // mapLock protects the volumes map for short lookups and metadata-only mutations. + // It must NOT be held during I/O operations (mount/unmount syscalls). + mapLock sync.RWMutex + // volLocks provides per-volume mutexes so that mount/unmount I/O for + // different volumes can proceed concurrently. + volLocks map[string]*sync.Mutex } // NewAmazonECSVolumePlugin initiates the volume drivers @@ -48,16 +53,28 @@ func NewAmazonECSVolumePlugin() *AmazonECSVolumePlugin { "efs": NewECSVolumeDriver(), "s3files": NewECSVolumeDriver(), }, - volumes: make(map[string]*types.Volume), - state: NewStateManager(), + volumes: make(map[string]*types.Volume), + state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } return plugin } +// getOrCreateVolLock returns the per-volume mutex, creating one if it doesn't exist. +// Caller must hold mapLock (at least RLock for read, Lock for create). +func (a *AmazonECSVolumePlugin) getOrCreateVolLock(name string) *sync.Mutex { + if mu, ok := a.volLocks[name]; ok { + return mu + } + mu := &sync.Mutex{} + a.volLocks[name] = mu + return mu +} + // LoadState loads past state information of the plugin func (a *AmazonECSVolumePlugin) LoadState() error { - a.lock.Lock() - defer a.lock.Unlock() + a.mapLock.Lock() + defer a.mapLock.Unlock() seelog.Info("Loading plugin state information") oldState := &VolumeState{} if !fileExists(PluginStateFileAbsPath) { @@ -96,6 +113,7 @@ func (a *AmazonECSVolumePlugin) LoadState() error { Mounts: vol.Mounts, } a.volumes[volName] = volume + a.getOrCreateVolLock(volName) voldriver.Setup(volName, volume) } a.state.VolState = oldState @@ -112,10 +130,11 @@ func (a *AmazonECSVolumePlugin) getVolumeDriver(driverType string) (driver.Volum return a.volumeDrivers[driverType], nil } -// Create implements Docker volume plugin's Create Method +// Create implements Docker volume plugin's Create Method. +// Create is metadata-only (no I/O), so the global lock is acceptable here. func (a *AmazonECSVolumePlugin) Create(r *volume.CreateRequest) error { - a.lock.Lock() - defer a.lock.Unlock() + a.mapLock.Lock() + defer a.mapLock.Unlock() seelog.Infof("Creating new volume %s", r.Name) _, ok := a.volumes[r.Name] @@ -162,6 +181,7 @@ func (a *AmazonECSVolumePlugin) Create(r *volume.CreateRequest) error { } // record the volume information a.volumes[r.Name] = vol + a.getOrCreateVolLock(r.Name) seelog.Infof("Saving state of new volume %s", r.Name) // save the state of new volume err = a.state.recordVolume(r.Name, vol) @@ -198,7 +218,8 @@ func deleteMountPath(path string) error { return os.Remove(path) } -// Mount implements Docker volume plugin's Mount Method +// Mount implements Docker volume plugin's Mount Method. +// Uses per-volume locking so that mounts for different volumes proceed concurrently. func (a *AmazonECSVolumePlugin) Mount(r *volume.MountRequest) (*volume.MountResponse, error) { seelog.Infof("Received mount request %+v", r) @@ -210,27 +231,30 @@ func (a *AmazonECSVolumePlugin) Mount(r *volume.MountRequest) (*volume.MountResp return nil, fmt.Errorf("no mount ID in the request") } - // Acquire write lock - a.lock.Lock() - defer a.lock.Unlock() - - // Find the volume + // Phase 1: short global lock to look up volume metadata and per-volume lock + a.mapLock.RLock() vol, ok := a.volumes[r.Name] if !ok { + a.mapLock.RUnlock() seelog.Errorf("Volume %s to mount is not found", r.Name) return nil, fmt.Errorf("volume %s not found", r.Name) } - - // Find the volume driver volDriver, err := a.getVolumeDriver(vol.Type) if err != nil { + a.mapLock.RUnlock() seelog.Errorf("Volume %s's driver type %s not supported: %v", r.Name, vol.Type, err) return nil, fmt.Errorf("Volume %s's driver type %s not supported: %w", r.Name, vol.Type, err) } if volDriver == nil { - // This case shouldn't happen normally + a.mapLock.RUnlock() return nil, fmt.Errorf("no volume driver found for type %s", vol.Type) } + volMu := a.getOrCreateVolLock(r.Name) + a.mapLock.RUnlock() + + // Phase 2: per-volume lock — only this volume is blocked, others proceed freely + volMu.Lock() + defer volMu.Unlock() // Mount the volume on the host if there are no active mounts for the volume. if len(vol.Mounts) == 0 { @@ -264,7 +288,8 @@ func (a *AmazonECSVolumePlugin) Mount(r *volume.MountRequest) (*volume.MountResp return &volume.MountResponse{Mountpoint: vol.Path}, nil } -// Unmount implements Docker volume plugin's Unmount Method +// Unmount implements Docker volume plugin's Unmount Method. +// Uses per-volume locking so that unmounts for different volumes proceed concurrently. func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { seelog.Infof("Received unmount request %+v", r) @@ -276,27 +301,30 @@ func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { return fmt.Errorf("no mount ID in the request") } - // Acquire write lock - a.lock.Lock() - defer a.lock.Unlock() - - // Find the volume + // Phase 1: short global lock to look up volume metadata and per-volume lock + a.mapLock.RLock() vol, ok := a.volumes[r.Name] if !ok { + a.mapLock.RUnlock() seelog.Errorf("Volume %s to unmount is not found", r.Name) return fmt.Errorf("volume %s not found", r.Name) } - - // Get the corresponding volume driver volDriver, err := a.getVolumeDriver(vol.Type) if err != nil { + a.mapLock.RUnlock() seelog.Errorf("Volume %s removal failure: %v", r.Name, err) return fmt.Errorf("volume %v of type %s is unsupported: %w", r.Name, vol.Type, err) } if volDriver == nil { - // this case should not happen normally + a.mapLock.RUnlock() return fmt.Errorf("no corresponding volume driver found for type %s", vol.Type) } + volMu := a.getOrCreateVolLock(r.Name) + a.mapLock.RUnlock() + + // Phase 2: per-volume lock + volMu.Lock() + defer volMu.Unlock() // Remove the mount from the volume seelog.Infof("Removing mount %s from volume %s", r.ID, r.Name) @@ -324,12 +352,13 @@ func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { return nil } -// Remove implements Docker volume plugin's Remove Method +// Remove implements Docker volume plugin's Remove Method. +// Uses global write lock since it mutates the volumes map. func (a *AmazonECSVolumePlugin) Remove(r *volume.RemoveRequest) error { seelog.Infof("Received Remove request %+v", r) - a.lock.Lock() - defer a.lock.Unlock() + a.mapLock.Lock() + defer a.mapLock.Unlock() seelog.Infof("Removing volume %s", r.Name) vol, ok := a.volumes[r.Name] @@ -362,6 +391,7 @@ func (a *AmazonECSVolumePlugin) Remove(r *volume.RemoveRequest) error { // remove the volume information delete(a.volumes, r.Name) + delete(a.volLocks, r.Name) // cleanup the volume's host mount path err = a.CleanupMountPath(vol.Path) if err != nil { @@ -378,8 +408,8 @@ func (a *AmazonECSVolumePlugin) Remove(r *volume.RemoveRequest) error { // List implements Docker volume plugin's List Method func (a *AmazonECSVolumePlugin) List() (*volume.ListResponse, error) { - a.lock.RLock() - defer a.lock.RUnlock() + a.mapLock.RLock() + defer a.mapLock.RUnlock() vols := make([]*volume.Volume, len(a.volumes)) i := 0 for volName := range a.volumes { @@ -396,8 +426,8 @@ func (a *AmazonECSVolumePlugin) List() (*volume.ListResponse, error) { // Get implements Docker volume plugin's Get Method func (a *AmazonECSVolumePlugin) Get(r *volume.GetRequest) (*volume.GetResponse, error) { - a.lock.RLock() - defer a.lock.RUnlock() + a.mapLock.RLock() + defer a.mapLock.RUnlock() vol, ok := a.volumes[r.Name] if !ok { return nil, fmt.Errorf("volume %s not found", r.Name) @@ -413,8 +443,8 @@ func (a *AmazonECSVolumePlugin) Get(r *volume.GetRequest) (*volume.GetResponse, // Path implements Docker volume plugin's Path Method func (a *AmazonECSVolumePlugin) Path(r *volume.PathRequest) (*volume.PathResponse, error) { - a.lock.RLock() - defer a.lock.RUnlock() + a.mapLock.RLock() + defer a.mapLock.RUnlock() vol, ok := a.volumes[r.Name] if !ok { seelog.Errorf("Could not find mount path for volume %s", r.Name) diff --git a/ecs-init/volumes/ecs_volume_plugin_test.go b/ecs-init/volumes/ecs_volume_plugin_test.go index e8dd3054839..90803a10ebd 100644 --- a/ecs-init/volumes/ecs_volume_plugin_test.go +++ b/ecs-init/volumes/ecs_volume_plugin_test.go @@ -14,7 +14,10 @@ package volumes import ( "errors" + "fmt" + "sync" "testing" + "time" "github.com/aws/amazon-ecs-agent/ecs-init/volumes/driver" mock_driver "github.com/aws/amazon-ecs-agent/ecs-init/volumes/driver/mock" @@ -77,8 +80,9 @@ func TestVolumeCreateHappyPath(t *testing.T) { volumeDrivers: map[string]driver.VolumeDriver{ "efs": NewTestVolumeDriver(), }, - volumes: make(map[string]*types.Volume), - state: NewStateManager(), + volumes: make(map[string]*types.Volume), + state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.CreateRequest{ Name: "vol", @@ -119,6 +123,7 @@ func TestVolumeCreateTargetSpecified(t *testing.T) { }, volumes: make(map[string]*types.Volume), state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.CreateRequest{ Name: "vol", @@ -154,6 +159,7 @@ func TestVolumeCreateSaveFailure(t *testing.T) { }, volumes: make(map[string]*types.Volume), state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.CreateRequest{ Name: "vol", @@ -209,6 +215,7 @@ func TestCreateNoVolumeType(t *testing.T) { }, volumes: make(map[string]*types.Volume), state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.CreateRequest{ Name: "vol", @@ -340,6 +347,7 @@ func TestVolumeRemoveHappyPath(t *testing.T) { volName: vol, }, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.RemoveRequest{Name: volName} removeMountPath = func(path string) error { @@ -378,6 +386,7 @@ func TestVolumeRemoveFailure(t *testing.T) { volName: vol, }, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } saveStateToDisk = func(b []byte) error { return nil @@ -397,6 +406,7 @@ func TestRemoveVolumeNotFound(t *testing.T) { }, volumes: map[string]*types.Volume{}, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.RemoveRequest{Name: "vol"} assert.Error(t, plugin.Remove(req), "expected error when volume to remove is not found") @@ -417,6 +427,7 @@ func TestRemoveVolumeDriverNotFound(t *testing.T) { volName: vol, }, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.RemoveRequest{Name: volName} assert.Error(t, plugin.Remove(req), "expected error when corresponding volume driver not found") @@ -443,6 +454,7 @@ func TestVolumeRemoveNoUnmountIfAlreadyUnmounted(t *testing.T) { volName: vol, }, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } saveStateToDisk = func(b []byte) error { return nil @@ -470,6 +482,7 @@ func TestVolumeRemoveMountPathFailure(t *testing.T) { volName: vol, }, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.RemoveRequest{Name: volName} removeMountPath = func(path string) error { @@ -502,6 +515,7 @@ func TestVolumeRemoveStateSaveFailure(t *testing.T) { volName: vol, }, state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.RemoveRequest{Name: volName} removeMountPath = func(path string) error { @@ -686,6 +700,7 @@ func TestPluginLoadState(t *testing.T) { }, volumes: make(map[string]*types.Volume), state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } fileExists = func(path string) bool { return true @@ -706,6 +721,7 @@ func TestPluginLoadState(t *testing.T) { func TestPluginNoStateFile(t *testing.T) { plugin := &AmazonECSVolumePlugin{ state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } fileExists = func(path string) bool { return false @@ -719,6 +735,7 @@ func TestPluginNoStateFile(t *testing.T) { func TestPluginInvalidState(t *testing.T) { plugin := &AmazonECSVolumePlugin{ state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } fileExists = func(path string) bool { return true @@ -740,6 +757,7 @@ func TestPluginEmptyState(t *testing.T) { }, volumes: make(map[string]*types.Volume), state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } fileExists = func(path string) bool { return true @@ -969,6 +987,7 @@ func TestPluginMount(t *testing.T) { volumes: tc.pluginVolumes, volumeDrivers: map[string]driver.VolumeDriver{driverTypeEFS: efsDriver}, state: pluginState, + volLocks: make(map[string]*sync.Mutex), } for volName, vol := range tc.pluginVolumes { pluginState.recordVolume(volName, vol) @@ -1164,6 +1183,7 @@ func TestPluginUnmount(t *testing.T) { volumes: tc.pluginVolumes, volumeDrivers: map[string]driver.VolumeDriver{driverTypeEFS: efsDriver}, state: pluginState, + volLocks: make(map[string]*sync.Mutex), } for volName, vol := range tc.pluginVolumes { pluginState.recordVolume(volName, vol) @@ -1202,6 +1222,7 @@ func TestCreate_S3FilesVolume(t *testing.T) { }, volumes: make(map[string]*types.Volume), state: NewStateManager(), + volLocks: make(map[string]*sync.Mutex), } req := &volume.CreateRequest{ Name: "s3vol", @@ -1230,3 +1251,75 @@ func TestCreate_S3FilesVolume(t *testing.T) { assert.Equal(t, VolumeMountPathPrefix+"s3vol", vol.Path) assert.NotEmpty(t, vol.CreatedAt) } + +// TestConcurrentMountsDifferentVolumes verifies that mount operations for different +// volumes can proceed concurrently rather than being serialized behind a single lock. +func TestConcurrentMountsDifferentVolumes(t *testing.T) { + const numVolumes = 5 + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + saveStateToDisk = func(b []byte) error { return nil } + defer func() { saveStateToDisk = saveState }() + + // Create a slow mock driver that simulates I/O latency + slowDriver := mock_driver.NewMockVolumeDriver(ctrl) + for i := 0; i < numVolumes; i++ { + volName := fmt.Sprintf("vol%d", i) + slowDriver.EXPECT(). + Create(gomock.Any()). + DoAndReturn(func(r *driver.CreateRequest) error { + time.Sleep(50 * time.Millisecond) + return nil + }) + _ = volName + } + + // Set up plugin with pre-created volumes + volumes := make(map[string]*types.Volume) + for i := 0; i < numVolumes; i++ { + volumes[fmt.Sprintf("vol%d", i)] = &types.Volume{ + Path: fmt.Sprintf("/mnt/vol%d", i), + Mounts: map[string]int{}, + Options: map[string]string{}, + } + } + + pluginState := NewStateManager() + plugin := &AmazonECSVolumePlugin{ + volumes: volumes, + volumeDrivers: map[string]driver.VolumeDriver{"efs": slowDriver}, + state: pluginState, + volLocks: make(map[string]*sync.Mutex), + } + for volName, vol := range volumes { + pluginState.recordVolume(volName, vol) + } + + // Launch concurrent mount requests + start := time.Now() + var wg sync.WaitGroup + errs := make([]error, numVolumes) + for i := 0; i < numVolumes; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + _, errs[idx] = plugin.Mount(&volume.MountRequest{ + Name: fmt.Sprintf("vol%d", idx), + ID: fmt.Sprintf("mount%d", idx), + }) + }(i) + } + wg.Wait() + elapsed := time.Since(start) + + for i, err := range errs { + assert.NoError(t, err, "mount vol%d should succeed", i) + } + + // With serial processing, 5 x 50ms = 250ms minimum. + // With concurrent processing, all should complete in ~50-100ms. + assert.Less(t, elapsed.Milliseconds(), int64(200), + "concurrent mounts should complete faster than serial execution") +} From 943fd4cd0add678e8fed246337b46251123795df Mon Sep 17 00:00:00 2001 From: taewankim <38234018+callaway12@users.noreply.github.com> Date: Thu, 2 Jul 2026 12:12:33 +0900 Subject: [PATCH 2/2] volumes: fix race conditions in per-volume locking and state management - Split getOrCreateVolLock() into getVolLock(), createVolLock(), and getOrCreateVolLock() with double-checked locking to eliminate map write under RLock - Extend StateManager lock to cover map mutation and marshaling in recordVolume()/removeVolume(), fixing concurrent map access race - Add per-volume lock coordination to Remove() to prevent conflicts with in-flight Mount/Unmount operations - Add re-check after acquiring per-volume lock in Mount/Unmount to handle concurrent removal - Replace wall-clock assertion with sync.WaitGroup barrier pattern in TestConcurrentMountsDifferentVolumes - Add TestConcurrentMountsSameVolume to verify single driver Create() call under concurrent mounts - Fix seelog.Errorf format string in Unmount (missing %v for error) --- ecs-init/volumes/ecs_volume_plugin.go | 117 ++++++++++++++++----- ecs-init/volumes/ecs_volume_plugin_test.go | 95 +++++++++++++---- ecs-init/volumes/state_manager.go | 13 ++- ecs-init/volumes/state_manager_test.go | 15 ++- 4 files changed, 186 insertions(+), 54 deletions(-) diff --git a/ecs-init/volumes/ecs_volume_plugin.go b/ecs-init/volumes/ecs_volume_plugin.go index 47c9fe71a8a..5c5dcd4cede 100644 --- a/ecs-init/volumes/ecs_volume_plugin.go +++ b/ecs-init/volumes/ecs_volume_plugin.go @@ -60,14 +60,37 @@ func NewAmazonECSVolumePlugin() *AmazonECSVolumePlugin { return plugin } -// getOrCreateVolLock returns the per-volume mutex, creating one if it doesn't exist. -// Caller must hold mapLock (at least RLock for read, Lock for create). +// getVolLock returns the per-volume mutex for an existing volume. +// Caller must hold mapLock (at least RLock). +// Returns nil if no lock exists for the given volume name. +func (a *AmazonECSVolumePlugin) getVolLock(name string) *sync.Mutex { + return a.volLocks[name] +} + +// createVolLock creates a per-volume mutex for the given volume name. +// Caller must hold mapLock for writing (Lock, not RLock). +func (a *AmazonECSVolumePlugin) createVolLock(name string) *sync.Mutex { + mu := &sync.Mutex{} + a.volLocks[name] = mu + return mu +} + +// getOrCreateVolLock returns the per-volume mutex, creating one atomically if needed. +// Safe to call without holding mapLock — uses double-checked locking internally. func (a *AmazonECSVolumePlugin) getOrCreateVolLock(name string) *sync.Mutex { - if mu, ok := a.volLocks[name]; ok { + a.mapLock.RLock() + mu := a.volLocks[name] + a.mapLock.RUnlock() + if mu != nil { return mu } - mu := &sync.Mutex{} - a.volLocks[name] = mu + a.mapLock.Lock() + mu = a.volLocks[name] + if mu == nil { + mu = &sync.Mutex{} + a.volLocks[name] = mu + } + a.mapLock.Unlock() return mu } @@ -113,7 +136,7 @@ func (a *AmazonECSVolumePlugin) LoadState() error { Mounts: vol.Mounts, } a.volumes[volName] = volume - a.getOrCreateVolLock(volName) + a.createVolLock(volName) voldriver.Setup(volName, volume) } a.state.VolState = oldState @@ -181,7 +204,7 @@ func (a *AmazonECSVolumePlugin) Create(r *volume.CreateRequest) error { } // record the volume information a.volumes[r.Name] = vol - a.getOrCreateVolLock(r.Name) + a.createVolLock(r.Name) seelog.Infof("Saving state of new volume %s", r.Name) // save the state of new volume err = a.state.recordVolume(r.Name, vol) @@ -231,7 +254,7 @@ func (a *AmazonECSVolumePlugin) Mount(r *volume.MountRequest) (*volume.MountResp return nil, fmt.Errorf("no mount ID in the request") } - // Phase 1: short global lock to look up volume metadata and per-volume lock + // Phase 1: short global lock to look up volume metadata a.mapLock.RLock() vol, ok := a.volumes[r.Name] if !ok { @@ -249,13 +272,26 @@ func (a *AmazonECSVolumePlugin) Mount(r *volume.MountRequest) (*volume.MountResp a.mapLock.RUnlock() return nil, fmt.Errorf("no volume driver found for type %s", vol.Type) } - volMu := a.getOrCreateVolLock(r.Name) + volMu := a.getVolLock(r.Name) a.mapLock.RUnlock() + if volMu == nil { + volMu = a.getOrCreateVolLock(r.Name) + } + // Phase 2: per-volume lock — only this volume is blocked, others proceed freely volMu.Lock() defer volMu.Unlock() + // Re-check that the volume still exists after acquiring the per-volume lock, + // as Remove() may have deleted it while we were waiting. + a.mapLock.RLock() + if _, ok := a.volumes[r.Name]; !ok { + a.mapLock.RUnlock() + return nil, fmt.Errorf("volume %s was removed", r.Name) + } + a.mapLock.RUnlock() + // Mount the volume on the host if there are no active mounts for the volume. if len(vol.Mounts) == 0 { seelog.Infof("Mounting volume %s as there are no existing mounts for it", r.Name) @@ -301,7 +337,7 @@ func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { return fmt.Errorf("no mount ID in the request") } - // Phase 1: short global lock to look up volume metadata and per-volume lock + // Phase 1: short global lock to look up volume metadata a.mapLock.RLock() vol, ok := a.volumes[r.Name] if !ok { @@ -319,13 +355,25 @@ func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { a.mapLock.RUnlock() return fmt.Errorf("no corresponding volume driver found for type %s", vol.Type) } - volMu := a.getOrCreateVolLock(r.Name) + volMu := a.getVolLock(r.Name) a.mapLock.RUnlock() + if volMu == nil { + volMu = a.getOrCreateVolLock(r.Name) + } + // Phase 2: per-volume lock volMu.Lock() defer volMu.Unlock() + // Re-check that the volume still exists after acquiring the per-volume lock. + a.mapLock.RLock() + if _, ok := a.volumes[r.Name]; !ok { + a.mapLock.RUnlock() + return fmt.Errorf("volume %s was removed", r.Name) + } + a.mapLock.RUnlock() + // Remove the mount from the volume seelog.Infof("Removing mount %s from volume %s", r.ID, r.Name) if exists := vol.RemoveMount(r.ID); !exists { @@ -345,7 +393,7 @@ func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { // Save state if err := a.state.recordVolume(r.Name, vol); err != nil { // State save failed, so roll back the changes made so far to make state consistent - seelog.Errorf("Error saving state of volume %s", r.Name, err) + seelog.Errorf("Error saving state of volume %s: %v", r.Name, err) } // All good @@ -353,52 +401,73 @@ func (a *AmazonECSVolumePlugin) Unmount(r *volume.UnmountRequest) error { } // Remove implements Docker volume plugin's Remove Method. -// Uses global write lock since it mutates the volumes map. +// Acquires the per-volume lock to coordinate with in-flight Mount/Unmount operations, +// then holds the global write lock to mutate the volumes map. func (a *AmazonECSVolumePlugin) Remove(r *volume.RemoveRequest) error { seelog.Infof("Received Remove request %+v", r) - a.mapLock.Lock() - defer a.mapLock.Unlock() - - seelog.Infof("Removing volume %s", r.Name) + // Phase 1: look up volume and per-volume lock under read lock + a.mapLock.RLock() vol, ok := a.volumes[r.Name] if !ok { + a.mapLock.RUnlock() seelog.Errorf("Volume %s to remove is not found", r.Name) return fmt.Errorf("volume %s not found", r.Name) } - - // get corresponding volume driver to unmount volDriver, err := a.getVolumeDriver(vol.Type) if err != nil { + a.mapLock.RUnlock() seelog.Errorf("Volume %s removal failure: %s", r.Name, err) return err } if volDriver == nil { - // this case should not happen normally + a.mapLock.RUnlock() return fmt.Errorf("no corresponding volume driver found for type %s", vol.Type) } + volMu := a.getVolLock(r.Name) + a.mapLock.RUnlock() + + // Phase 2: acquire per-volume lock to wait for in-flight Mount/Unmount to finish + if volMu != nil { + volMu.Lock() + defer volMu.Unlock() + } + + // Phase 3: re-check existence under read lock + a.mapLock.RLock() + if _, ok := a.volumes[r.Name]; !ok { + a.mapLock.RUnlock() + seelog.Errorf("Volume %s to remove is not found", r.Name) + return fmt.Errorf("volume %s not found", r.Name) + } + a.mapLock.RUnlock() + + seelog.Infof("Removing volume %s", r.Name) // Although unmounts are handled by Unmount method, unmount the volume if it's still // mounted. This is mainly to unmount volumes created by an older version of the // plugin in which unmounts were not handled by Unmount method. + // This runs outside mapLock so other volumes are not blocked during slow I/O. if volDriver.IsMounted(r.Name) { - seelog.Infof("Volume %s is currently mounted, unmouting it", r.Name) + seelog.Infof("Volume %s is currently mounted, unmounting it", r.Name) if err := volDriver.Remove(&driver.RemoveRequest{Name: r.Name}); err != nil { seelog.Errorf("Volume %s removal failure: %v", r.Name, err) return err } } - // remove the volume information + // Phase 4: acquire global write lock only for map mutation + a.mapLock.Lock() delete(a.volumes, r.Name) delete(a.volLocks, r.Name) - // cleanup the volume's host mount path + a.mapLock.Unlock() + + // Cleanup and state save outside global lock err = a.CleanupMountPath(vol.Path) if err != nil { seelog.Errorf("Cleaning mount path failed for volume %s: %v", r.Name, err) } seelog.Infof("Saving state after removing volume %s", r.Name) - // remove the state of deleted volume err = a.state.removeVolume(r.Name) if err != nil { seelog.Errorf("Error saving state after removing volume %s: %v", r.Name, err) diff --git a/ecs-init/volumes/ecs_volume_plugin_test.go b/ecs-init/volumes/ecs_volume_plugin_test.go index 90803a10ebd..f629591329e 100644 --- a/ecs-init/volumes/ecs_volume_plugin_test.go +++ b/ecs-init/volumes/ecs_volume_plugin_test.go @@ -17,7 +17,6 @@ import ( "fmt" "sync" "testing" - "time" "github.com/aws/amazon-ecs-agent/ecs-init/volumes/driver" mock_driver "github.com/aws/amazon-ecs-agent/ecs-init/volumes/driver/mock" @@ -1254,6 +1253,7 @@ func TestCreate_S3FilesVolume(t *testing.T) { // TestConcurrentMountsDifferentVolumes verifies that mount operations for different // volumes can proceed concurrently rather than being serialized behind a single lock. +// Uses a barrier (sync.WaitGroup) instead of wall-clock timing to prove concurrency. func TestConcurrentMountsDifferentVolumes(t *testing.T) { const numVolumes = 5 @@ -1263,20 +1263,21 @@ func TestConcurrentMountsDifferentVolumes(t *testing.T) { saveStateToDisk = func(b []byte) error { return nil } defer func() { saveStateToDisk = saveState }() - // Create a slow mock driver that simulates I/O latency - slowDriver := mock_driver.NewMockVolumeDriver(ctrl) - for i := 0; i < numVolumes; i++ { - volName := fmt.Sprintf("vol%d", i) - slowDriver.EXPECT(). - Create(gomock.Any()). - DoAndReturn(func(r *driver.CreateRequest) error { - time.Sleep(50 * time.Millisecond) - return nil - }) - _ = volName - } + // Barrier: each Create() signals arrival, then waits for all to arrive. + // If mounts were serialized, the second goroutine would never enter Create() + // until the first returns — so the barrier would deadlock (caught by test timeout). + var entered sync.WaitGroup + entered.Add(numVolumes) + + mockDriver := mock_driver.NewMockVolumeDriver(ctrl) + mockDriver.EXPECT(). + Create(gomock.Any()). + DoAndReturn(func(r *driver.CreateRequest) error { + entered.Done() + entered.Wait() + return nil + }).Times(numVolumes) - // Set up plugin with pre-created volumes volumes := make(map[string]*types.Volume) for i := 0; i < numVolumes; i++ { volumes[fmt.Sprintf("vol%d", i)] = &types.Volume{ @@ -1289,7 +1290,7 @@ func TestConcurrentMountsDifferentVolumes(t *testing.T) { pluginState := NewStateManager() plugin := &AmazonECSVolumePlugin{ volumes: volumes, - volumeDrivers: map[string]driver.VolumeDriver{"efs": slowDriver}, + volumeDrivers: map[string]driver.VolumeDriver{"efs": mockDriver}, state: pluginState, volLocks: make(map[string]*sync.Mutex), } @@ -1297,8 +1298,6 @@ func TestConcurrentMountsDifferentVolumes(t *testing.T) { pluginState.recordVolume(volName, vol) } - // Launch concurrent mount requests - start := time.Now() var wg sync.WaitGroup errs := make([]error, numVolumes) for i := 0; i < numVolumes; i++ { @@ -1312,14 +1311,66 @@ func TestConcurrentMountsDifferentVolumes(t *testing.T) { }(i) } wg.Wait() - elapsed := time.Since(start) for i, err := range errs { assert.NoError(t, err, "mount vol%d should succeed", i) } +} - // With serial processing, 5 x 50ms = 250ms minimum. - // With concurrent processing, all should complete in ~50-100ms. - assert.Less(t, elapsed.Milliseconds(), int64(200), - "concurrent mounts should complete faster than serial execution") +// TestConcurrentMountsSameVolume verifies that concurrent Mount() calls for the +// same volume result in exactly one driver Create() call. +func TestConcurrentMountsSameVolume(t *testing.T) { + const ( + volName = "shared-vol" + numMounts = 5 + ) + + ctrl := gomock.NewController(t) + defer ctrl.Finish() + + saveStateToDisk = func(b []byte) error { return nil } + defer func() { saveStateToDisk = saveState }() + + mockDriver := mock_driver.NewMockVolumeDriver(ctrl) + mockDriver.EXPECT(). + Create(&driver.CreateRequest{ + Name: volName, + Path: "/mnt/shared", + Options: map[string]string{}, + }). + Return(nil). + Times(1) + + pluginState := NewStateManager() + vol := &types.Volume{ + Path: "/mnt/shared", + Mounts: map[string]int{}, + Options: map[string]string{}, + } + plugin := &AmazonECSVolumePlugin{ + volumes: map[string]*types.Volume{volName: vol}, + volumeDrivers: map[string]driver.VolumeDriver{"efs": mockDriver}, + state: pluginState, + volLocks: make(map[string]*sync.Mutex), + } + pluginState.recordVolume(volName, vol) + + var wg sync.WaitGroup + errs := make([]error, numMounts) + for i := 0; i < numMounts; i++ { + wg.Add(1) + go func(idx int) { + defer wg.Done() + _, errs[idx] = plugin.Mount(&volume.MountRequest{ + Name: volName, + ID: fmt.Sprintf("mount%d", idx), + }) + }(i) + } + wg.Wait() + + for i, err := range errs { + assert.NoError(t, err, "mount %d should succeed", i) + } + assert.Equal(t, numMounts, len(vol.Mounts), "all mounts should be recorded") } diff --git a/ecs-init/volumes/state_manager.go b/ecs-init/volumes/state_manager.go index 20bb63a8023..4bae60880a6 100644 --- a/ecs-init/volumes/state_manager.go +++ b/ecs-init/volumes/state_manager.go @@ -66,7 +66,9 @@ func NewStateManager() *StateManager { } func (s *StateManager) recordVolume(volName string, vol *types.Volume) error { - // Copy the mounts so that the map is not shared + s.lock.Lock() + defer s.lock.Unlock() + mountsCopy := map[string]int{} for k, v := range vol.Mounts { mountsCopy[k] = v @@ -83,15 +85,16 @@ func (s *StateManager) recordVolume(volName string, vol *types.Volume) error { } func (s *StateManager) removeVolume(volName string) error { + s.lock.Lock() + defer s.lock.Unlock() + delete(s.VolState.Volumes, volName) return s.save() } -// saves volume state to the file at path +// save marshals and persists the volume state to disk. +// Caller must hold s.lock. func (s *StateManager) save() error { - s.lock.Lock() - defer s.lock.Unlock() - b, err := json.MarshalIndent(s.VolState, "", "\t") if err != nil { return fmt.Errorf("marshal data failed: %v", err) diff --git a/ecs-init/volumes/state_manager_test.go b/ecs-init/volumes/state_manager_test.go index 63773553b47..f3d17349dbe 100644 --- a/ecs-init/volumes/state_manager_test.go +++ b/ecs-init/volumes/state_manager_test.go @@ -41,7 +41,10 @@ func TestSaveStateSuccess(t *testing.T) { defer func() { saveStateToDisk = saveState }() - assert.NoError(t, s.save()) + s.lock.Lock() + err := s.save() + s.lock.Unlock() + assert.NoError(t, err) } func TestSaveStateToDisk(t *testing.T) { @@ -56,7 +59,10 @@ func TestSaveStateToDisk(t *testing.T) { defer func() { saveStateToDisk = saveState }() - assert.NoError(t, s.save()) + s.lock.Lock() + err := s.save() + s.lock.Unlock() + assert.NoError(t, err) } func TestSaveStateToDiskFail(t *testing.T) { @@ -71,7 +77,10 @@ func TestSaveStateToDiskFail(t *testing.T) { defer func() { saveStateToDisk = saveState }() - assert.Error(t, s.save()) + s.lock.Lock() + err := s.save() + s.lock.Unlock() + assert.Error(t, err) } func TestLoadStateSuccess(t *testing.T) {