From 0e9d206867bde2e40bccd5130414da620f2c2855 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Sun, 19 Mar 2023 21:21:30 +0800 Subject: [PATCH] fix(region,host): block jobs progress refactor Signed-off-by: wanyaoqi --- pkg/apis/compute/guests.go | 5 + pkg/compute/models/guest_actions.go | 1 + .../tasks/guest_change_disk_storage_task.go | 36 +-- .../guestman/guesthandlers/guesthandler.go | 16 +- pkg/hostman/guestman/guestman.go | 4 + pkg/hostman/guestman/guesttasks.go | 255 +++++++++++++----- pkg/hostman/guestman/qemu-kvm.go | 19 +- pkg/hostman/monitor/hmp.go | 2 +- pkg/hostman/monitor/monitor.go | 30 +-- pkg/hostman/monitor/qmp.go | 46 +--- 10 files changed, 244 insertions(+), 170 deletions(-) diff --git a/pkg/apis/compute/guests.go b/pkg/apis/compute/guests.go index 1a26a66dd1..26339826c7 100644 --- a/pkg/apis/compute/guests.go +++ b/pkg/apis/compute/guests.go @@ -856,6 +856,7 @@ type ServerChangeStorageInternalInput struct { ServerChangeStorageInput Disks []string `json:"disks"` GuestRunning bool `json:"guest_running"` + DiskCount int `json:"disk_count"` } type ServerChangeDiskStorageInput struct { @@ -871,6 +872,10 @@ type ServerChangeDiskStorageInternalInput struct { DiskFormat string `json:"disk_format"` GuestRunning bool `json:"guest_running"` TargetDiskDesc *GuestdiskJsonDesc `json:"target_disk_desc"` + + // clone progress + CompletedDiskCount int `json:"completed_disk_count"` + CloneDiskCount int `json:"disk_count"` } type ServerSetExtraOptionInput struct { diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 32b2b10daf..6cd9d18d55 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -5504,6 +5504,7 @@ func (self *SGuest) StartGuestChangeStorageTask(ctx context.Context, userCred mc params := api.ServerChangeStorageInternalInput{ ServerChangeStorageInput: *input, Disks: disks, + DiskCount: len(disks), GuestRunning: self.Status == api.VM_RUNNING, } reason := fmt.Sprintf("Change guest disks storage to %s", input.TargetStorageId) diff --git a/pkg/compute/tasks/guest_change_disk_storage_task.go b/pkg/compute/tasks/guest_change_disk_storage_task.go index 8008d3fd34..8094ee1fd5 100644 --- a/pkg/compute/tasks/guest_change_disk_storage_task.go +++ b/pkg/compute/tasks/guest_change_disk_storage_task.go @@ -86,21 +86,9 @@ func (t *GuestChangeDiskStorageTask) ChangeDiskStorage(ctx context.Context, gues t.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("GetTargetDisk error: %v", err))) return } - sourceDisk, err := t.GetSourceDisk() - if err != nil { - t.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("GetSourceDisk error: %v", err))) - return - } // set target disk's status to clone targetDisk.SetStatus(t.GetUserCred(), api.DISK_CLONE, "") - - err = sourceDisk.SetMetadata(ctx, api.DISK_CLONE_TASK_ID, t.GetId(), t.GetUserCred()) - if err != nil { - t.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("SetMetadata clone task id: %v", err))) - return - } - log.Infof("ChangeDiskStorage guest running is %v", input.GuestRunning) if input.GuestRunning { t.SetStage("OnDiskLiveChangeStorageReady", nil) @@ -189,12 +177,6 @@ func (t *GuestChangeDiskStorageTask) OnDiskChangeStorageComplete(ctx context.Con return } - err = srcDisk.SetMetadata(ctx, api.DISK_CLONE_TASK_ID, "", t.GetUserCred()) - if err != nil { - t.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("SetMetadata clone task id: %v", err))) - return - } - input, err := t.GetInputParams() if err != nil { t.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("GetInputParams error: %v", err))) @@ -333,11 +315,6 @@ type GuestChangeDisksStorageTask struct { SGuestBaseTask } -func (t *GuestChangeDisksStorageTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) { - guest := obj.(*models.SGuest) - t.ChangeDiskStorage(ctx, guest, nil) -} - func (t *GuestChangeDisksStorageTask) GetInputParams() (*api.ServerChangeStorageInternalInput, error) { input := new(api.ServerChangeStorageInternalInput) err := t.GetParams().Unmarshal(input) @@ -370,6 +347,11 @@ func (t *GuestChangeDisksStorageTask) OnGuestSyncStatusFailed(ctx context.Contex t.TaskFailed(ctx, guest, data) } +func (t *GuestChangeDisksStorageTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) { + guest := obj.(*models.SGuest) + t.ChangeDiskStorage(ctx, guest, nil) +} + func (t *GuestChangeDisksStorageTask) ChangeDiskStorage(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { input, err := t.GetInputParams() if err != nil { @@ -420,9 +402,11 @@ func (t *GuestChangeDisksStorageTask) CreateTargetDisk(ctx context.Context, gues TargetStorageId: storage.Id, KeepOriginDisk: input.KeepOriginDisk, }, - StorageId: srcDisk.StorageId, - TargetDiskId: targetDisk.GetId(), - GuestRunning: input.GuestRunning, + StorageId: srcDisk.StorageId, + TargetDiskId: targetDisk.GetId(), + GuestRunning: input.GuestRunning, + CloneDiskCount: input.DiskCount, + CompletedDiskCount: input.DiskCount - len(input.Disks) - 1, } if err := guest.StartChangeDiskStorageTask(ctx, t.UserCred, internalInput, t.Id); err != nil { diff --git a/pkg/hostman/guestman/guesthandlers/guesthandler.go b/pkg/hostman/guestman/guesthandlers/guesthandler.go index e63df5465d..4f1f210f51 100644 --- a/pkg/hostman/guestman/guesthandlers/guesthandler.go +++ b/pkg/hostman/guestman/guesthandlers/guesthandler.go @@ -689,13 +689,15 @@ func formatCloneDiskParams(sid string, body jsonutils.JSONObject) (*guestman.SSt } params := &guestman.SStorageCloneDisk{ - ServerId: sid, - SourceStorage: srcStorage, - SourceDisk: srcDisk, - TargetStorage: targetStorage, - TargetDiskId: input.TargetDiskId, - DiskFormat: input.DiskFormat, - TargetDiskDesc: input.TargetDiskDesc, + ServerId: sid, + SourceStorage: srcStorage, + SourceDisk: srcDisk, + TargetStorage: targetStorage, + TargetDiskId: input.TargetDiskId, + DiskFormat: input.DiskFormat, + TargetDiskDesc: input.TargetDiskDesc, + CompletedDiskCount: input.CompletedDiskCount, + CloneDiskCount: input.CloneDiskCount, } return params, nil } diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index 4a9d0b54f9..13c5d8bc01 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -1338,6 +1338,10 @@ type SStorageCloneDisk struct { TargetDiskId string DiskFormat string TargetDiskDesc *compute.GuestdiskJsonDesc + + // clone progress + CompletedDiskCount int + CloneDiskCount int } func (m *SGuestManager) StorageCloneDisk(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index 253059b2c8..6789e40bb9 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/pkg/hostman/guestman/guesttasks.go @@ -1501,14 +1501,105 @@ func (s *SGuestResumeTask) removeStatefile() { go s.CleanStatefiles() } +type IGuestBlockProgressTask interface { + OnGetBlockJobs(jobs []monitor.BlockJob) + StreamingDiskCompletedCount() int + StreamingDiskCount() int +} + +type SGuestBlockProgressBaseTask struct { + *SKVMGuestInstance + + ctx context.Context + + totalSizeMb int + completedSizeMb int + + c chan struct{} + jobs map[string]monitor.BlockJob + task IGuestBlockProgressTask +} + +func NewGuestBlockProgressBaseTask( + ctx context.Context, guest *SKVMGuestInstance, blkTask IGuestBlockProgressTask, +) *SGuestBlockProgressBaseTask { + return &SGuestBlockProgressBaseTask{ + SKVMGuestInstance: guest, + ctx: ctx, + task: blkTask, + jobs: map[string]monitor.BlockJob{}, + } +} + +func (s *SGuestBlockProgressBaseTask) startWaitBlockJob(res string) { + s.c = make(chan struct{}) + for { + select { + case <-s.c: + s.c = nil + return + case <-time.After(time.Second * 10): + s.Monitor.GetBlockJobs(s.onGetBlockJobs) + } + } +} + +func (s *SGuestBlockProgressBaseTask) onGetBlockJobs(jobs []monitor.BlockJob) { + if len(jobs) == 0 && s.c != nil { + close(s.c) + s.c = nil + } + for i := range jobs { + job := jobs[i] + _job, ok := s.jobs[job.Device] + if !ok { + job.CalcOffset(0) + s.jobs[job.Device] = job + continue + } + if _job.Status == "ready" { + delete(s.jobs, _job.Device) + continue + } + job.Start, job.Now = _job.Start, _job.Now + job.CalcOffset(_job.Offset) + s.jobs[job.Device] = job + } + mbps, progress := 0.0, 0.0 + totalSize, totalOffset := int64(1), int64(0) + for _, job := range s.jobs { + mbps += job.SpeedMbps + totalSize += job.Len + totalOffset += job.Offset + } + if len(s.jobs) == 0 && len(jobs) == 0 { + progress = 100.0 + } else { + progress = float64(totalOffset) / float64(totalSize) * 100 + } + + diskCount := s.task.StreamingDiskCount() + if diskCount > 0 { + progress = float64(s.task.StreamingDiskCompletedCount())/float64(diskCount)*100.0 + 1.0/float64(diskCount)*progress + } + hostutils.UpdateServerProgress(context.Background(), s.GetId(), progress, mbps) + s.task.OnGetBlockJobs(jobs) +} + +func (s *SGuestBlockProgressBaseTask) cancelWaitBlockJobs() { + if s.c != nil { + close(s.c) + s.c = nil + } +} + /** * GuestStreamDisksTask **/ type SGuestStreamDisksTask struct { - *SKVMGuestInstance + *SGuestBlockProgressBaseTask - ctx context.Context callback func() disksIdx []int @@ -1517,12 +1608,12 @@ type SGuestStreamDisksTask struct { } func NewGuestStreamDisksTask(ctx context.Context, guest *SKVMGuestInstance, callback func(), disksIdx []int) *SGuestStreamDisksTask { - return &SGuestStreamDisksTask{ - SKVMGuestInstance: guest, - ctx: ctx, - callback: callback, - disksIdx: disksIdx, + task := &SGuestStreamDisksTask{ + callback: callback, + disksIdx: disksIdx, } + task.SGuestBlockProgressBaseTask = NewGuestBlockProgressBaseTask(ctx, guest, task) + return task } func (s *SGuestStreamDisksTask) Start() { @@ -1532,7 +1623,7 @@ func (s *SGuestStreamDisksTask) Start() { func (s *SGuestStreamDisksTask) onInitCheckStreamJobs(jobs int) { if jobs > 0 { log.Warningf("GuestStreamDisksTask: duplicate block streaming???") - s.startWaitBlockStream("") + s.startWaitBlockJob("") } else if jobs == 0 { s.startBlockStreaming() } @@ -1576,35 +1667,28 @@ func (s *SGuestStreamDisksTask) startDoBlockStream() { if len(s.streamDevs) > 0 { dev := s.streamDevs[0] s.streamDevs = s.streamDevs[1:] - s.Monitor.BlockStream(dev, len(s.disksIdx)-len(s.streamDevs), len(s.disksIdx), s.startWaitBlockStream) + s.Monitor.BlockStream(dev, s.startWaitBlockJob) } else { s.taskComplete() } } -func (s *SGuestStreamDisksTask) startWaitBlockStream(res string) { - log.Infof("Block stream command res %s: %q", s.GetName(), res) - s.c = make(chan struct{}) - for { - select { - case <-s.c: - s.c = nil - return - case <-time.After(time.Second * 10): - s.Monitor.GetBlockJobCounts(s.checkStreamJobs) - } - } +func (s *SGuestStreamDisksTask) StreamingDiskCompletedCount() int { + return len(s.disksIdx) - len(s.streamDevs) - 1 } -func (s *SGuestStreamDisksTask) checkStreamJobs(jobs int) { - if jobs == 0 && s.c != nil { - close(s.c) +func (s *SGuestStreamDisksTask) StreamingDiskCount() int { + return len(s.disksIdx) +} + +func (s *SGuestStreamDisksTask) OnGetBlockJobs(jobs []monitor.BlockJob) { + if len(jobs) == 0 { + s.cancelWaitBlockJobs() if s.streamDevs == nil { s.checkBlockDrives() } else { s.startDoBlockStream() } - } } @@ -1612,25 +1696,6 @@ func (s *SGuestStreamDisksTask) taskComplete() { hostutils.UpdateServerProgress(context.Background(), s.Id, 100.0, 0.0) s.SyncStatus("") - // XXX: region disk post-migrate not implement - - // disks, _ := s.Desc.GetArray("disks") - // var needSync = fale - // for i, disk := range disks { - // if disk.Contains("url") && disk.Contains("path") { - // diskId, _ := disk.GetString("disk_id") - // targetStroageId, _ := disk.GetString("target_storage_id") - // params := jsonutils.NewDict() - // params.Set("storage_id", jsonutils.NewString(targetStroageId)) - // modules.Disks.PerformAction(hostutils.GetComputeSession(context.Background()), - // diskId, "post-migrate", params) - // needSync = true - // } - // } - // if needSync { - // s.SaveLiveDesc(s.Desc) - // } - if s.callback != nil { s.callback() } @@ -2518,18 +2583,20 @@ func (task *SCancelBlockJobs) taskComplete() { } type SGuestStorageCloneDiskTask struct { - *SKVMGuestInstance + *SGuestBlockProgressBaseTask - ctx context.Context - params *SStorageCloneDisk + started bool + diskIndex int + resp *hostapi.ServerCloneDiskFromStorageResponse + params *SStorageCloneDisk } func NewGuestStorageCloneDiskTask(ctx context.Context, guest *SKVMGuestInstance, params *SStorageCloneDisk) *SGuestStorageCloneDiskTask { - return &SGuestStorageCloneDiskTask{ - SKVMGuestInstance: guest, - ctx: ctx, - params: params, + task := &SGuestStorageCloneDiskTask{ + params: params, } + task.SGuestBlockProgressBaseTask = NewGuestBlockProgressBaseTask(ctx, guest, task) + return task } func (t *SGuestStorageCloneDiskTask) Start(guestRunning bool) { @@ -2576,24 +2643,13 @@ func (t *SGuestStorageCloneDiskTask) Start(guestRunning bool) { if !guestRunning { hostutils.TaskComplete(t.ctx, jsonutils.Marshal(resp)) return - } - - onDriveMirror := func(res string) { - if len(res) > 0 { - hostutils.TaskFailed( - t.ctx, fmt.Sprintf("Clone disk %s to storage %s drive mirror failed %s", - t.params.SourceDisk.GetPath(), t.params.TargetStorage.GetId(), res), - ) - } else { - hostutils.TaskComplete(t.ctx, jsonutils.Marshal(resp)) - timeutils2.AddTimeout(time.Second*3, func() { - t.SyncStatus("drive mirror started") - }) - } + } else { + t.resp = resp + t.diskIndex = diskIndex } t.Monitor.DriveMirror( - onDriveMirror, + t.onDriveMirror, fmt.Sprintf("drive_%d", diskIndex), targetDisk.GetPath(), "full", @@ -2603,6 +2659,73 @@ func (t *SGuestStorageCloneDiskTask) Start(guestRunning bool) { ) } +func (t *SGuestStorageCloneDiskTask) onDriveMirror(res string) { + if len(res) > 0 { + hostutils.TaskFailed( + t.ctx, fmt.Sprintf("Clone disk %s to storage %s drive mirror failed %s", + t.params.SourceDisk.GetPath(), t.params.TargetStorage.GetId(), res), + ) + } else { + t.started = true + hostutils.TaskComplete(t.ctx, jsonutils.Marshal(t.resp)) + go t.startWaitBlockJob(res) + timeutils2.AddTimeout(time.Second*3, func() { + t.SyncStatus("drive mirror started") + }) + } +} + +func (t *SGuestStorageCloneDiskTask) OnGetBlockJobs(jobs []monitor.BlockJob) { + if len(jobs) == 0 { + if !t.started { + targetDisk, err := t.params.TargetStorage.GetDiskById(t.params.TargetDiskId) + if err != nil { + hostutils.TaskFailed( + t.ctx, fmt.Sprintf("Failed get target disk %s %s", t.params.TargetDiskId, err), + ) + return + } + targetDiskFormat, err := targetDisk.GetFormat() + if err != nil { + hostutils.TaskFailed( + t.ctx, fmt.Sprintf("Failed get target disk format %s %s", t.params.TargetDiskId, err), + ) + return + } + t.Monitor.DriveMirror( + t.onDriveMirror, + fmt.Sprintf("drive_%d", t.diskIndex), + targetDisk.GetPath(), + "full", + targetDiskFormat, + true, + false, + ) + } else { + hostutils.TaskFailed(t.ctx, fmt.Sprintf("Disk %s Block job not found", t.params.SourceDisk.GetId())) + } + return + } + for i := range jobs { + if jobs[i].Status == "ready" && jobs[i].Device == fmt.Sprintf("drive_%d", t.diskIndex) { + t.cancelWaitBlockJobs() + params := jsonutils.NewDict() + params.Set("block_jobs_ready", jsonutils.JSONTrue) + hostutils.TaskComplete(t.ctx, params) + break + } + } + +} + +func (t *SGuestStorageCloneDiskTask) StreamingDiskCompletedCount() int { + return t.params.CompletedDiskCount +} + +func (t *SGuestStorageCloneDiskTask) StreamingDiskCount() int { + return t.params.CloneDiskCount +} + type SGuestLiveChangeDisk struct { *SKVMGuestInstance diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index 4c7041ada2..1d970651ea 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -512,6 +512,9 @@ func (s *SKVMGuestInstance) asyncScriptStart(ctx context.Context, params interfa err = s.updateGuestDesc() } if err != nil { + if ctx != nil && len(appctx.AppContextTaskId(ctx)) >= 0 { + hostutils.TaskFailed(ctx, fmt.Sprintf("Async start server failed: %s", err)) + } return nil, errors.Wrap(err, "asyncScriptStart init desc") } @@ -914,16 +917,6 @@ func (s *SKVMGuestInstance) eventBlockJobReady(event *monitor.Event) { } } } - } else { - params := jsonutils.NewDict() - params.Set("disk_id", jsonutils.NewString(diskId)) - _, err = modules.Servers.PerformAction( - hostutils.GetComputeSession(context.Background()), - s.GetId(), "block-mirror-ready", params, - ) - if err != nil { - log.Errorf("Server %s perform block-mirror-ready got error %s", s.GetId(), err) - } } } @@ -1236,7 +1229,7 @@ func (s *SKVMGuestInstance) SlaveDisksBlockStream() error { for i := 0; i < len(disks); i++ { diskIndex := disks[i].Index drive := fmt.Sprintf("drive_%d", diskIndex) - s.Monitor.BlockStream(drive, 0, 0, func(res string) { + s.Monitor.BlockStream(drive, func(res string) { errChan <- res }) if errStr := <-errChan; len(errStr) > 0 { @@ -1482,7 +1475,9 @@ type guestStartTask struct { } func (t *guestStartTask) Run() { - t.s.asyncScriptStart(t.ctx, t.params) + if _, err := t.s.asyncScriptStart(t.ctx, t.params); err != nil { + log.Errorf("failed asyncScriptStart %s: %s", t.s.GetId(), err) + } } func (t *guestStartTask) Dump() string { diff --git a/pkg/hostman/monitor/hmp.go b/pkg/hostman/monitor/hmp.go index bb24f42014..5ec403dd5d 100644 --- a/pkg/hostman/monitor/hmp.go +++ b/pkg/hostman/monitor/hmp.go @@ -423,7 +423,7 @@ func (m *HmpMonitor) DriveBackup(callback StringCallback, drive, target, syncMod m.Query(cmd, callback) } -func (m *HmpMonitor) BlockStream(drive string, _, _ int, callback StringCallback) { +func (m *HmpMonitor) BlockStream(drive string, callback StringCallback) { var ( speed = 500 // limit 500 MB/s cmd = fmt.Sprintf("block_stream %s %d", drive, speed) diff --git a/pkg/hostman/monitor/monitor.go b/pkg/hostman/monitor/monitor.go index d192232347..dd45360e89 100644 --- a/pkg/hostman/monitor/monitor.go +++ b/pkg/hostman/monitor/monitor.go @@ -44,10 +44,10 @@ type BlockJob struct { Device string Speed int64 - start time.Time - preOffset int64 - now time.Time - speedMbps float64 + Start time.Time + PreOffset int64 + Now time.Time + SpeedMbps float64 } type QemuBlock struct { @@ -165,22 +165,22 @@ func (self blockSizeByte) String() string { return fmt.Sprintf("%d", int64(self)) } -func (self *BlockJob) PreOffset(preOffset int64) { - if self.start.IsZero() { - self.start = time.Now() - self.now = time.Now() - self.preOffset = preOffset +func (self *BlockJob) CalcOffset(preOffset int64) { + if self.Start.IsZero() { + self.Start = time.Now() + self.Now = time.Now() + self.PreOffset = preOffset return } - second := time.Now().Sub(self.now).Seconds() + second := time.Now().Sub(self.Now).Seconds() if second > 0 { speed := float64(self.Offset-preOffset) / second - self.speedMbps = speed / 1024 / 1024 - avgSpeed := float64(self.Offset) / time.Now().Sub(self.start).Seconds() + self.SpeedMbps = speed / 1024 / 1024 + avgSpeed := float64(self.Offset) / time.Now().Sub(self.Start).Seconds() log.Infof(`[%s / %s] server %s block job for %s speed: %s/s(avg: %s/s)`, blockSizeByte(self.Offset).String(), blockSizeByte(self.Len).String(), self.server, self.Device, blockSizeByte(speed).String(), blockSizeByte(avgSpeed).String()) } - self.preOffset = preOffset - self.now = time.Now() + self.PreOffset = preOffset + self.Now = time.Now() return } @@ -222,7 +222,7 @@ type Monitor interface { DeviceAdd(dev string, params map[string]string, callback StringCallback) XBlockdevChange(parent, node, child string, callback StringCallback) - BlockStream(drive string, idx, blkCnt int, callback StringCallback) + BlockStream(drive string, callback StringCallback) DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool) DriveBackup(callback StringCallback, drive, target, syncMode, format string) BlockJobComplete(drive string, cb StringCallback) diff --git a/pkg/hostman/monitor/qmp.go b/pkg/hostman/monitor/qmp.go index dc39cd67a1..296a41fc46 100644 --- a/pkg/hostman/monitor/qmp.go +++ b/pkg/hostman/monitor/qmp.go @@ -112,11 +112,6 @@ type QmpMonitor struct { commandQueue []*Command callbackQueue []qmpMonitorCallBack jobs map[string]BlockJob - - blkStream struct { - idx int - blkCnt int - } } func NewQmpMonitor(server, sid string, OnMonitorDisConnect, OnMonitorTimeout MonitorErrorFunc, @@ -819,41 +814,8 @@ func (m *QmpMonitor) blockJobs(res *Response) ([]BlockJob, error) { } log.Debugf("blockJobs response %s", ret) jobs := []BlockJob{} - ret.Unmarshal(&jobs) - defer func() { - mbps, progress := 0.0, 0.0 - totalSize, totalOffset := int64(1), int64(0) - for _, job := range m.jobs { - mbps += job.speedMbps - totalSize += job.Len - totalOffset += job.Offset - } - if len(m.jobs) == 0 && len(jobs) == 0 { - progress = 100.0 - } else { - progress = float64(totalOffset) / float64(totalSize) * 100 - } - if m.blkStream.blkCnt > 0 { - progress = float64(m.blkStream.idx-1)/float64(m.blkStream.blkCnt)*100.0 + 1.0/float64(m.blkStream.blkCnt)*progress - } - hostutils.UpdateServerProgress(context.Background(), m.sid, progress, mbps) - }() - for i := range jobs { - job := jobs[i] - job.server = m.server - _job, ok := m.jobs[job.Device] - if !ok { - job.PreOffset(0) - m.jobs[job.Device] = job - continue - } - if _job.Status == "ready" { - delete(m.jobs, _job.Device) - continue - } - job.start, job.now = _job.start, _job.now - job.PreOffset(_job.Offset) - m.jobs[job.Device] = job + if err = ret.Unmarshal(&jobs); err != nil { + return nil, err } return jobs, nil } @@ -941,7 +903,7 @@ func (m *QmpMonitor) DriveBackup(callback StringCallback, drive, target, syncMod m.Query(cmd, cb) } -func (m *QmpMonitor) BlockStream(drive string, idx, blkCnt int, callback StringCallback) { +func (m *QmpMonitor) BlockStream(drive string, callback StringCallback) { var ( speed = 5 * 100 * 1024 * 1024 // limit 500 MB/s cb = func(res *Response) { @@ -955,8 +917,6 @@ func (m *QmpMonitor) BlockStream(drive string, idx, blkCnt int, callback StringC }, } ) - m.blkStream.idx = idx - m.blkStream.blkCnt = blkCnt m.Query(cmd, cb) }