Merge pull request #5404 from alexlarsson/dm-new-metadata

Make devicemapper backend able to support multiple processes

daemon/graphdriver/devmapper/deviceset.go

@@ -11,7 +11,6 @@ import (
 	"path"
 	"path/filepath"
 	"strconv"
-	"strings"
 	"sync"
 	"syscall"
 	"time"
@@ -62,8 +61,7 @@ type DeviceSet struct {
 	devicePrefix     string
 	TransactionId    uint64
 	NewTransactionId uint64
-	nextFreeDevice   int
-	sawBusy          bool
+	nextDeviceId     int
 }
 
 type DiskUsage struct {
@@ -109,7 +107,19 @@ func (devices *DeviceSet) loopbackDir() string {
 	return path.Join(devices.root, "devicemapper")
 }
 
-func (devices *DeviceSet) jsonFile() string {
+func (devices *DeviceSet) metadataDir() string {
+	return path.Join(devices.root, "metadata")
+}
+
+func (devices *DeviceSet) metadataFile(info *DevInfo) string {
+	file := info.Hash
+	if file == "" {
+		file = "base"
+	}
+	return path.Join(devices.metadataDir(), file)
+}
+
+func (devices *DeviceSet) oldMetadataFile() string {
 	return path.Join(devices.loopbackDir(), "json")
 }
 
@@ -159,26 +169,24 @@ func (devices *DeviceSet) ensureImage(name string, size int64) (string, error) {
 	return filename, nil
 }
 
-func (devices *DeviceSet) allocateDeviceId() int {
-	// TODO: Add smarter reuse of deleted devices
-	id := devices.nextFreeDevice
-	devices.nextFreeDevice = devices.nextFreeDevice + 1
-	return id
-}
-
 func (devices *DeviceSet) allocateTransactionId() uint64 {
 	devices.NewTransactionId = devices.NewTransactionId + 1
 	return devices.NewTransactionId
 }
 
-func (devices *DeviceSet) saveMetadata() error {
-	devices.devicesLock.Lock()
-	jsonData, err := json.Marshal(devices.MetaData)
-	devices.devicesLock.Unlock()
+func (devices *DeviceSet) removeMetadata(info *DevInfo) error {
+	if err := osRemoveAll(devices.metadataFile(info)); err != nil {
+		return fmt.Errorf("Error removing metadata file %s: %s", devices.metadataFile(info), err)
+	}
+	return nil
+}
+
+func (devices *DeviceSet) saveMetadata(info *DevInfo) error {
+	jsonData, err := json.Marshal(info)
 	if err != nil {
 		return fmt.Errorf("Error encoding metadata to json: %s", err)
 	}
-	tmpFile, err := ioutil.TempFile(filepath.Dir(devices.jsonFile()), ".json")
+	tmpFile, err := ioutil.TempFile(devices.metadataDir(), ".tmp")
 	if err != nil {
 		return fmt.Errorf("Error creating metadata file: %s", err)
 	}
@@ -196,7 +204,7 @@ func (devices *DeviceSet) saveMetadata() error {
 	if err := tmpFile.Close(); err != nil {
 		return fmt.Errorf("Error closing metadata file %s: %s", tmpFile.Name(), err)
 	}
-	if err := osRename(tmpFile.Name(), devices.jsonFile()); err != nil {
+	if err := osRename(tmpFile.Name(), devices.metadataFile(info)); err != nil {
 		return fmt.Errorf("Error committing metadata file %s: %s", tmpFile.Name(), err)
 	}
 
@@ -214,7 +222,12 @@ func (devices *DeviceSet) lookupDevice(hash string) (*DevInfo, error) {
 	defer devices.devicesLock.Unlock()
 	info := devices.Devices[hash]
 	if info == nil {
-		return nil, fmt.Errorf("Unknown device %s", hash)
+		info = devices.loadMetadata(hash)
+		if info == nil {
+			return nil, fmt.Errorf("Unknown device %s", hash)
+		}
+
+		devices.Devices[hash] = info
 	}
 	return info, nil
 }
@@ -234,7 +247,7 @@ func (devices *DeviceSet) registerDevice(id int, hash string, size uint64) (*Dev
 	devices.Devices[hash] = info
 	devices.devicesLock.Unlock()
 
-	if err := devices.saveMetadata(); err != nil {
+	if err := devices.saveMetadata(info); err != nil {
 		// Try to remove unused device
 		devices.devicesLock.Lock()
 		delete(devices.Devices, hash)
@@ -269,9 +282,7 @@ func (devices *DeviceSet) createFilesystem(info *DevInfo) error {
 	return nil
 }
 
-func (devices *DeviceSet) loadMetaData() error {
-	utils.Debugf("loadMetadata()")
-	defer utils.Debugf("loadMetadata END")
+func (devices *DeviceSet) initMetaData() error {
 	_, _, _, params, err := getStatus(devices.getPoolName())
 	if err != nil {
 		utils.Debugf("\n--->Err: %s\n", err)
@@ -284,39 +295,64 @@ func (devices *DeviceSet) loadMetaData() error {
 	}
 	devices.NewTransactionId = devices.TransactionId
 
-	jsonData, err := ioutil.ReadFile(devices.jsonFile())
+	// Migrate old metadatafile
+
+	jsonData, err := ioutil.ReadFile(devices.oldMetadataFile())
 	if err != nil && !osIsNotExist(err) {
 		utils.Debugf("\n--->Err: %s\n", err)
 		return err
 	}
 
-	devices.MetaData.Devices = make(map[string]*DevInfo)
 	if jsonData != nil {
-		if err := json.Unmarshal(jsonData, &devices.MetaData); err != nil {
+		m := MetaData{Devices: make(map[string]*DevInfo)}
+
+		if err := json.Unmarshal(jsonData, &m); err != nil {
 			utils.Debugf("\n--->Err: %s\n", err)
 			return err
 		}
-	}
 
-	for hash, d := range devices.Devices {
-		d.Hash = hash
-		d.devices = devices
+		for hash, info := range m.Devices {
+			info.Hash = hash
 
-		if d.DeviceId >= devices.nextFreeDevice {
-			devices.nextFreeDevice = d.DeviceId + 1
+			// If the transaction id is larger than the actual one we lost the device due to some crash
+			if info.TransactionId <= devices.TransactionId {
+				devices.saveMetadata(info)
+			}
 		}
-
-		// If the transaction id is larger than the actual one we lost the device due to some crash
-		if d.TransactionId > devices.TransactionId {
-			utils.Debugf("Removing lost device %s with id %d", hash, d.TransactionId)
-			delete(devices.Devices, hash)
+		if err := osRename(devices.oldMetadataFile(), devices.oldMetadataFile()+".migrated"); err != nil {
+			return err
 		}
+
 	}
+
 	return nil
 }
 
+func (devices *DeviceSet) loadMetadata(hash string) *DevInfo {
+	info := &DevInfo{Hash: hash, devices: devices}
+
+	jsonData, err := ioutil.ReadFile(devices.metadataFile(info))
+	if err != nil {
+		return nil
+	}
+
+	if err := json.Unmarshal(jsonData, &info); err != nil {
+		return nil
+	}
+
+	fmt.Printf("Loaded metadata %v\n", info)
+
+	// If the transaction id is larger than the actual one we lost the device due to some crash
+	if info.TransactionId > devices.TransactionId {
+		return nil
+	}
+
+	return info
+}
+
 func (devices *DeviceSet) setupBaseImage() error {
 	oldInfo, _ := devices.lookupDevice("")
+	utils.Debugf("oldInfo: %p", oldInfo)
 	if oldInfo != nil && oldInfo.Initialized {
 		return nil
 	}
@@ -331,14 +367,17 @@ func (devices *DeviceSet) setupBaseImage() error {
 
 	utils.Debugf("Initializing base device-manager snapshot")
 
-	id := devices.allocateDeviceId()
+	id := devices.nextDeviceId
 
 	// Create initial device
-	if err := createDevice(devices.getPoolDevName(), id); err != nil {
+	if err := createDevice(devices.getPoolDevName(), &id); err != nil {
 		utils.Debugf("\n--->Err: %s\n", err)
 		return err
 	}
 
+	// Ids are 24bit, so wrap around
+	devices.nextDeviceId = (id + 1) & 0xffffff
+
 	utils.Debugf("Registering base device (id %v) with FS size %v", id, DefaultBaseFsSize)
 	info, err := devices.registerDevice(id, "", DefaultBaseFsSize)
 	if err != nil {
@@ -360,7 +399,7 @@ func (devices *DeviceSet) setupBaseImage() error {
 	}
 
 	info.Initialized = true
-	if err = devices.saveMetadata(); err != nil {
+	if err = devices.saveMetadata(info); err != nil {
 		info.Initialized = false
 		utils.Debugf("\n--->Err: %s\n", err)
 		return err
@@ -388,10 +427,6 @@ func (devices *DeviceSet) log(level int, file string, line int, dmError int, mes
 		return // Ignore _LOG_DEBUG
 	}
 
-	if strings.Contains(message, "busy") {
-		devices.sawBusy = true
-	}
-
 	utils.Debugf("libdevmapper(%d): %s:%d (%d) %s", level, file, line, dmError, message)
 }
 
@@ -472,29 +507,7 @@ func (devices *DeviceSet) ResizePool(size int64) error {
 func (devices *DeviceSet) initDevmapper(doInit bool) error {
 	logInit(devices)
 
-	// Make sure the sparse images exist in <root>/devicemapper/data and
-	// <root>/devicemapper/metadata
-
-	hasData := devices.hasImage("data")
-	hasMetadata := devices.hasImage("metadata")
-
-	if !doInit && !hasData {
-		return errors.New("Loopback data file not found")
-	}
-
-	if !doInit && !hasMetadata {
-		return errors.New("Loopback metadata file not found")
-	}
-
-	createdLoopback := !hasData || !hasMetadata
-	data, err := devices.ensureImage("data", DefaultDataLoopbackSize)
-	if err != nil {
-		utils.Debugf("Error device ensureImage (data): %s\n", err)
-		return err
-	}
-	metadata, err := devices.ensureImage("metadata", DefaultMetaDataLoopbackSize)
-	if err != nil {
-		utils.Debugf("Error device ensureImage (metadata): %s\n", err)
+	if err := osMkdirAll(devices.metadataDir(), 0700); err != nil && !osIsExist(err) {
 		return err
 	}
 
@@ -527,10 +540,38 @@ func (devices *DeviceSet) initDevmapper(doInit bool) error {
 	// so we add this badhack to make sure it closes itself
 	setCloseOnExec("/dev/mapper/control")
 
+	// Make sure the sparse images exist in <root>/devicemapper/data and
+	// <root>/devicemapper/metadata
+
+	createdLoopback := false
+
 	// If the pool doesn't exist, create it
 	if info.Exists == 0 {
 		utils.Debugf("Pool doesn't exist. Creating it.")
 
+		hasData := devices.hasImage("data")
+		hasMetadata := devices.hasImage("metadata")
+
+		if !doInit && !hasData {
+			return errors.New("Loopback data file not found")
+		}
+
+		if !doInit && !hasMetadata {
+			return errors.New("Loopback metadata file not found")
+		}
+
+		createdLoopback = !hasData || !hasMetadata
+		data, err := devices.ensureImage("data", DefaultDataLoopbackSize)
+		if err != nil {
+			utils.Debugf("Error device ensureImage (data): %s\n", err)
+			return err
+		}
+		metadata, err := devices.ensureImage("metadata", DefaultMetaDataLoopbackSize)
+		if err != nil {
+			utils.Debugf("Error device ensureImage (metadata): %s\n", err)
+			return err
+		}
+
 		dataFile, err := attachLoopDevice(data)
 		if err != nil {
 			utils.Debugf("\n--->Err: %s\n", err)
@@ -552,9 +593,9 @@ func (devices *DeviceSet) initDevmapper(doInit bool) error {
 	}
 
 	// If we didn't just create the data or metadata image, we need to
-	// load the metadata from the existing file.
+	// load the transaction id and migrate old metadata
 	if !createdLoopback {
-		if err = devices.loadMetaData(); err != nil {
+		if err = devices.initMetaData(); err != nil {
 			utils.Debugf("\n--->Err: %s\n", err)
 			return err
 		}
@@ -587,13 +628,16 @@ func (devices *DeviceSet) AddDevice(hash, baseHash string) error {
 		return fmt.Errorf("device %s already exists", hash)
 	}
 
-	deviceId := devices.allocateDeviceId()
+	deviceId := devices.nextDeviceId
 
-	if err := devices.createSnapDevice(devices.getPoolDevName(), deviceId, baseInfo.Name(), baseInfo.DeviceId); err != nil {
+	if err := createSnapDevice(devices.getPoolDevName(), &deviceId, baseInfo.Name(), baseInfo.DeviceId); err != nil {
 		utils.Debugf("Error creating snap device: %s\n", err)
 		return err
 	}
 
+	// Ids are 24bit, so wrap around
+	devices.nextDeviceId = (deviceId + 1) & 0xffffff
+
 	if _, err := devices.registerDevice(deviceId, hash, baseInfo.Size); err != nil {
 		deleteDevice(devices.getPoolDevName(), deviceId)
 		utils.Debugf("Error registering device: %s\n", err)
@@ -620,14 +664,6 @@ func (devices *DeviceSet) deleteDevice(info *DevInfo) error {
 		}
 	}
 
-	if info.Initialized {
-		info.Initialized = false
-		if err := devices.saveMetadata(); err != nil {
-			utils.Debugf("Error saving meta data: %s\n", err)
-			return err
-		}
-	}
-
 	if err := deleteDevice(devices.getPoolDevName(), info.DeviceId); err != nil {
 		utils.Debugf("Error deleting device: %s\n", err)
 		return err
@@ -638,11 +674,11 @@ func (devices *DeviceSet) deleteDevice(info *DevInfo) error {
 	delete(devices.Devices, info.Hash)
 	devices.devicesLock.Unlock()
 
-	if err := devices.saveMetadata(); err != nil {
+	if err := devices.removeMetadata(info); err != nil {
 		devices.devicesLock.Lock()
 		devices.Devices[info.Hash] = info
 		devices.devicesLock.Unlock()
-		utils.Debugf("Error saving meta data: %s\n", err)
+		utils.Debugf("Error removing meta data: %s\n", err)
 		return err
 	}
 
@@ -711,12 +747,11 @@ func (devices *DeviceSet) removeDeviceAndWait(devname string) error {
 	var err error
 
 	for i := 0; i < 1000; i++ {
-		devices.sawBusy = false
 		err = removeDevice(devname)
 		if err == nil {
 			break
 		}
-		if !devices.sawBusy {
+		if err != ErrBusy {
 			return err
 		}
 
@@ -886,7 +921,7 @@ func (devices *DeviceSet) MountDevice(hash, path, mountLabel string) error {
 	info.mountCount = 1
 	info.mountPath = path
 
-	return devices.setInitialized(info)
+	return nil
 }
 
 func (devices *DeviceSet) UnmountDevice(hash string) error {
@@ -937,14 +972,6 @@ func (devices *DeviceSet) HasDevice(hash string) bool {
 	return info != nil
 }
 
-func (devices *DeviceSet) HasInitializedDevice(hash string) bool {
-	devices.Lock()
-	defer devices.Unlock()
-
-	info, _ := devices.lookupDevice(hash)
-	return info != nil && info.Initialized
-}
-
 func (devices *DeviceSet) HasActivatedDevice(hash string) bool {
 	info, _ := devices.lookupDevice(hash)
 	if info == nil {
@@ -961,17 +988,6 @@ func (devices *DeviceSet) HasActivatedDevice(hash string) bool {
 	return devinfo != nil && devinfo.Exists != 0
 }
 
-func (devices *DeviceSet) setInitialized(info *DevInfo) error {
-	info.Initialized = true
-	if err := devices.saveMetadata(); err != nil {
-		info.Initialized = false
-		utils.Debugf("\n--->Err: %s\n", err)
-		return err
-	}
-
-	return nil
-}
-
 func (devices *DeviceSet) List() []string {
 	devices.Lock()
 	defer devices.Unlock()

daemon/graphdriver/devmapper/devmapper.go

@@ -62,6 +62,10 @@ var (
 	ErrInvalidAddNode         = errors.New("Invalide AddNoce type")
 	ErrGetLoopbackBackingFile = errors.New("Unable to get loopback backing file")
 	ErrLoopbackSetCapacity    = errors.New("Unable set loopback capacity")
+	ErrBusy                   = errors.New("Device is Busy")
+
+	dmSawBusy  bool
+	dmSawExist bool
 )
 
 type (
@@ -464,23 +468,33 @@ func resumeDevice(name string) error {
 	return nil
 }
 
-func createDevice(poolName string, deviceId int) error {
-	utils.Debugf("[devmapper] createDevice(poolName=%v, deviceId=%v)", poolName, deviceId)
-	task, err := createTask(DeviceTargetMsg, poolName)
-	if task == nil {
-		return err
-	}
+func createDevice(poolName string, deviceId *int) error {
+	utils.Debugf("[devmapper] createDevice(poolName=%v, deviceId=%v)", poolName, *deviceId)
 
-	if err := task.SetSector(0); err != nil {
-		return fmt.Errorf("Can't set sector")
-	}
+	for {
+		task, err := createTask(DeviceTargetMsg, poolName)
+		if task == nil {
+			return err
+		}
 
-	if err := task.SetMessage(fmt.Sprintf("create_thin %d", deviceId)); err != nil {
-		return fmt.Errorf("Can't set message")
-	}
+		if err := task.SetSector(0); err != nil {
+			return fmt.Errorf("Can't set sector")
+		}
 
-	if err := task.Run(); err != nil {
-		return fmt.Errorf("Error running createDevice")
+		if err := task.SetMessage(fmt.Sprintf("create_thin %d", *deviceId)); err != nil {
+			return fmt.Errorf("Can't set message")
+		}
+
+		dmSawExist = false
+		if err := task.Run(); err != nil {
+			if dmSawExist {
+				// Already exists, try next id
+				*deviceId++
+				continue
+			}
+			return fmt.Errorf("Error running createDevice")
+		}
+		break
 	}
 	return nil
 }
@@ -512,7 +526,11 @@ func removeDevice(name string) error {
 	if task == nil {
 		return err
 	}
+	dmSawBusy = false
 	if err = task.Run(); err != nil {
+		if dmSawBusy {
+			return ErrBusy
+		}
 		return fmt.Errorf("Error running removeDevice")
 	}
 	return nil
@@ -546,7 +564,7 @@ func activateDevice(poolName string, name string, deviceId int, size uint64) err
 	return nil
 }
 
-func (devices *DeviceSet) createSnapDevice(poolName string, deviceId int, baseName string, baseDeviceId int) error {
+func createSnapDevice(poolName string, deviceId *int, baseName string, baseDeviceId int) error {
 	devinfo, _ := getInfo(baseName)
 	doSuspend := devinfo != nil && devinfo.Exists != 0
 
@@ -556,33 +574,44 @@ func (devices *DeviceSet) createSnapDevice(poolName string, deviceId int, baseNa
 		}
 	}
 
-	task, err := createTask(DeviceTargetMsg, poolName)
-	if task == nil {
-		if doSuspend {
-			resumeDevice(baseName)
+	for {
+		task, err := createTask(DeviceTargetMsg, poolName)
+		if task == nil {
+			if doSuspend {
+				resumeDevice(baseName)
+			}
+			return err
 		}
-		return err
-	}
 
-	if err := task.SetSector(0); err != nil {
-		if doSuspend {
-			resumeDevice(baseName)
+		if err := task.SetSector(0); err != nil {
+			if doSuspend {
+				resumeDevice(baseName)
+			}
+			return fmt.Errorf("Can't set sector")
 		}
-		return fmt.Errorf("Can't set sector")
-	}
 
-	if err := task.SetMessage(fmt.Sprintf("create_snap %d %d", deviceId, baseDeviceId)); err != nil {
-		if doSuspend {
-			resumeDevice(baseName)
+		if err := task.SetMessage(fmt.Sprintf("create_snap %d %d", *deviceId, baseDeviceId)); err != nil {
+			if doSuspend {
+				resumeDevice(baseName)
+			}
+			return fmt.Errorf("Can't set message")
 		}
-		return fmt.Errorf("Can't set message")
-	}
 
-	if err := task.Run(); err != nil {
-		if doSuspend {
-			resumeDevice(baseName)
+		dmSawExist = false
+		if err := task.Run(); err != nil {
+			if dmSawExist {
+				// Already exists, try next id
+				*deviceId++
+				continue
+			}
+
+			if doSuspend {
+				resumeDevice(baseName)
+			}
+			return fmt.Errorf("Error running DeviceCreate (createSnapDevice)")
 		}
-		return fmt.Errorf("Error running DeviceCreate (createSnapDevice)")
+
+		break
 	}
 
 	if doSuspend {

daemon/graphdriver/devmapper/devmapper_log.go

@@ -4,12 +4,27 @@ package devmapper
 
 import "C"
 
+import (
+	"strings"
+)
+
 // Due to the way cgo works this has to be in a separate file, as devmapper.go has
 // definitions in the cgo block, which is incompatible with using "//export"
 
 //export DevmapperLogCallback
 func DevmapperLogCallback(level C.int, file *C.char, line C.int, dm_errno_or_class C.int, message *C.char) {
+	msg := C.GoString(message)
+	if level < 7 {
+		if strings.Contains(msg, "busy") {
+			dmSawBusy = true
+		}
+
+		if strings.Contains(msg, "File exists") {
+			dmSawExist = true
+		}
+	}
+
 	if dmLogger != nil {
-		dmLogger.log(int(level), C.GoString(file), int(line), int(dm_errno_or_class), C.GoString(message))
+		dmLogger.log(int(level), C.GoString(file), int(line), int(dm_errno_or_class), msg)
 	}
 }

daemon/graphdriver/devmapper/driver_test.go

@@ -825,10 +825,6 @@ func TestGetReturnsValidDevice(t *testing.T) {
 	if !d.HasActivatedDevice("1") {
 		t.Fatalf("Expected id 1 to be activated")
 	}
-
-	if !d.HasInitializedDevice("1") {
-		t.Fatalf("Expected id 1 to be initialized")
-	}
 }
 
 func TestDriverGetSize(t *testing.T) {