enhance: add volume lock

Signed-off-by: Eric Li <lcy041536@gmail.com>

storage/volume/core.go

@@ -5,6 +5,7 @@ import (
 	"reflect"
 	"strings"
 
+	"github.com/alibaba/pouch/pkg/kmutex"
 	metastore "github.com/alibaba/pouch/pkg/meta"
 	"github.com/alibaba/pouch/storage/volume/driver"
 	volerr "github.com/alibaba/pouch/storage/volume/error"
@@ -18,11 +19,15 @@ import (
 type Core struct {
 	Config
 	store *metastore.Store
+	lock  *kmutex.KMutex
 }
 
 // NewCore returns Core struct instance with volume config.
 func NewCore(cfg Config) (*Core, error) {
-	c := &Core{Config: cfg}
+	c := &Core{
+		Config: cfg,
+		lock:   kmutex.New(),
+	}
 
 	if cfg.DriverAlias != "" {
 		parts := strings.Split(cfg.DriverAlias, ";")
@@ -60,6 +65,14 @@ func NewCore(cfg Config) (*Core, error) {
 
 // GetVolume return a volume's info with specified name, If not errors.
 func (c *Core) GetVolume(id types.VolumeID) (*types.Volume, error) {
+	c.lock.Lock(id.Name)
+	defer c.lock.Unlock(id.Name)
+
+	return getVolume(id)
+}
+
+// getVolume return a volume's info with specified name, If not errors.
+func (c *Core) getVolume(id types.VolumeID) (*types.Volume, error) {
 	ctx := driver.Contexts()
 
 	// first, try to get volume from local store.
@@ -125,7 +138,7 @@ func (c *Core) GetVolume(id types.VolumeID) (*types.Volume, error) {
 
 // ExistVolume return 'true' if volume be found and not errors.
 func (c *Core) ExistVolume(id types.VolumeID) (bool, error) {
-	_, err := c.GetVolume(id)
+	_, err := c.getVolume(id)
 	if err != nil {
 		if ec, ok := err.(volerr.CoreError); ok && ec.IsVolumeNotFound() {
 			return false, nil
@@ -137,6 +150,9 @@ func (c *Core) ExistVolume(id types.VolumeID) (bool, error) {
 
 // CreateVolume use to create a volume, if failed, will return error info.
 func (c *Core) CreateVolume(id types.VolumeID) (*types.Volume, error) {
+	c.lock.Lock(id.Name)
+	defer c.lock.Unlock(id.Name)
+
 	exist, err := c.ExistVolume(id)
 	if err != nil {
 		return nil, err
@@ -265,6 +281,9 @@ func (c *Core) ListVolumeName(labels map[string]string) ([]string, error) {
 
 // RemoveVolume remove volume from storage and meta information, if not success return error.
 func (c *Core) RemoveVolume(id types.VolumeID) error {
+	c.lock.Lock(id.Name)
+	defer c.lock.Unlock(id.Name)
+
 	v, dv, err := c.GetVolumeDriver(id)
 	if err != nil {
 		return errors.Wrap(err, "Remove volume: "+id.String())
@@ -285,6 +304,9 @@ func (c *Core) RemoveVolume(id types.VolumeID) error {
 
 // VolumePath return the path of volume on node host.
 func (c *Core) VolumePath(id types.VolumeID) (string, error) {
+	c.lock.Lock(id.Name)
+	defer c.lock.Unlock(id.Name)
+
 	v, dv, err := c.GetVolumeDriver(id)
 	if err != nil {
 		return "", errors.Wrap(err, fmt.Sprintf("Get volume: %s path", id.String()))
@@ -295,7 +317,7 @@ func (c *Core) VolumePath(id types.VolumeID) (string, error) {
 
 // GetVolumeDriver return the backend driver and volume with specified volume's id.
 func (c *Core) GetVolumeDriver(id types.VolumeID) (*types.Volume, driver.Driver, error) {
-	v, err := c.GetVolume(id)
+	v, err := c.getVolume(id)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -308,6 +330,9 @@ func (c *Core) GetVolumeDriver(id types.VolumeID) (*types.Volume, driver.Driver,
 
 // AttachVolume to enable a volume on local host.
 func (c *Core) AttachVolume(id types.VolumeID, extra map[string]string) (*types.Volume, error) {
+	c.lock.Lock(id.Name)
+	defer c.lock.Unlock(id.Name)
+
 	v, dv, err := c.GetVolumeDriver(id)
 	if err != nil {
 		return nil, err
@@ -336,6 +361,9 @@ func (c *Core) AttachVolume(id types.VolumeID, extra map[string]string) (*types.
 
 // DetachVolume to disable a volume on local host.
 func (c *Core) DetachVolume(id types.VolumeID, extra map[string]string) (*types.Volume, error) {
+	c.lock.Lock(id.Name)
+	defer c.lock.Unlock(id.Name)
+
 	v, dv, err := c.GetVolumeDriver(id)
 	if err != nil {
 		return nil, err