opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-10-30 18:23:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(region,host): block jobs progress refactor

Browse Source 
Signed-off-by: wanyaoqi <d3lx.yq@gmail.com>
This commit is contained in:
wanyaoqi
2023-03-19 21:21:30 +08:00
parent 41c13e4da4
commit 0e9d206867
10 changed files with 244 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/compute/guests.go
+5
View File
@@ -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 {
pkg/compute/models/guest_actions.go
+1
View File
@@ -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)
pkg/compute/tasks/guest_change_disk_storage_task.go
+10 -26
View File
@@ -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 {
pkg/hostman/guestman/guesthandlers/guesthandler.go
+9 -7
View File
@@ -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
}
pkg/hostman/guestman/guestman.go
+4
View File
@@ -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) {
pkg/hostman/guestman/guesttasks.go
+189 -66
View File
@@ -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
pkg/hostman/guestman/qemu-kvm.go
+7 -12
View File
@@ -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 {
pkg/hostman/monitor/hmp.go
+1 -1
View File
@@ -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)
pkg/hostman/monitor/monitor.go
+15 -15
View File
@@ -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)
pkg/hostman/monitor/qmp.go
+3 -43
View File
@@ -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)
}