From a73e530cc15e3fa3a01e5643cc6128df30cf7dd7 Mon Sep 17 00:00:00 2001 From: ioito Date: Sat, 22 Jan 2022 15:48:18 +0800 Subject: [PATCH] feat(host): transfer sparse file over network --- cmd/climc/shell/compute/disks.go | 121 ++++++++++ pkg/hostman/downloader/downloader.go | 76 +++--- pkg/hostman/downloader/downloadhandler.go | 23 +- pkg/hostman/downloader/guest_downloader.go | 4 +- pkg/hostman/downloader/image_downloader.go | 4 +- .../downloader/imagecache_downloader.go | 4 +- pkg/hostman/downloader/snapshot_downloader.go | 4 +- .../storageman/remotefile/remotefile.go | 52 ++-- pkg/util/pb/doc.go | 1 + pkg/util/pb/pb.go | 134 +++++++++++ pkg/util/sparsefile/doc.go | 1 + pkg/util/sparsefile/hole_other.go | 26 ++ pkg/util/sparsefile/hole_unix.go | 60 +++++ pkg/util/sparsefile/sparse.go | 224 ++++++++++++++++++ 14 files changed, 645 insertions(+), 89 deletions(-) create mode 100644 pkg/util/pb/doc.go create mode 100644 pkg/util/pb/pb.go create mode 100644 pkg/util/sparsefile/doc.go create mode 100644 pkg/util/sparsefile/hole_other.go create mode 100644 pkg/util/sparsefile/hole_unix.go create mode 100644 pkg/util/sparsefile/sparse.go diff --git a/cmd/climc/shell/compute/disks.go b/cmd/climc/shell/compute/disks.go index 6c370496fc..a07046242c 100644 --- a/cmd/climc/shell/compute/disks.go +++ b/cmd/climc/shell/compute/disks.go @@ -15,11 +15,26 @@ package compute import ( + "compress/zlib" + "context" + "fmt" + "io" + "net/http" + "os" + "strconv" + "time" + + "github.com/cheggaaa/pb/v3" + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/pkg/errors" "yunion.io/x/onecloud/pkg/mcclient" modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute" "yunion.io/x/onecloud/pkg/mcclient/options" + "yunion.io/x/onecloud/pkg/util/httputils" + "yunion.io/x/onecloud/pkg/util/sparsefile" ) func init() { @@ -309,4 +324,110 @@ func init() { return nil }) + type DiskDownloadOptions struct { + ID string `help:"ID or name of disk" json:"-"` + Compress bool + Sparse bool + + Timeout int `help:"Timeout hours for download" default:"5"` + Debug bool + + FILE string + } + R(&DiskDownloadOptions{}, "disk-download", "Download disk from host", func(s *mcclient.ClientSession, args *DiskDownloadOptions) error { + disk, err := modules.Disks.GetById(s, args.ID, nil) + if err != nil { + return err + } + storageId, _ := disk.GetString("storage_id") + storage, err := modules.Storages.GetById(s, storageId, nil) + if err != nil { + return err + } + + hostsInfo := []struct { + Id string + }{} + storage.Unmarshal(&hostsInfo, "hosts") + header := http.Header{} + header.Set("X-Auth-Token", s.GetToken().GetTokenString()) + if args.Compress { + header.Set("X-Compress-Content", "zlib") + } + if args.Sparse { + header.Set("X-Sparse-Content", "true") + } + + client := httputils.GetTimeoutClient(time.Hour * time.Duration(args.Timeout)) + + for _, host := range hostsInfo { + host, err := modules.Hosts.GetById(s, host.Id, nil) + if err != nil { + return err + } + managerUri, _ := host.GetString("manager_uri") + if len(managerUri) == 0 { + continue + } + url := fmt.Sprintf("%s/download/disks/%s/%s", managerUri, storageId, args.ID) + resp, err := httputils.Request(client, context.Background(), httputils.GET, url, header, nil, args.Debug) + if err != nil { + log.Errorf("request %s error: %v", err) + continue + } + defer resp.Body.Close() + + totalSize, _ := strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64) + sparseHeader, _ := strconv.ParseInt(resp.Header.Get("X-Sparse-Header"), 10, 64) + + fi, err := os.Create(args.FILE) + if err != nil { + return errors.Wrapf(err, "os.Create(%s)", args.FILE) + } + defer fi.Close() + + var reader = resp.Body + + if args.Compress { + zlibRC, err := zlib.NewReader(resp.Body) + if err != nil { + return errors.Wrapf(err, "zlib.NewReader") + } + defer zlibRC.Close() + reader = zlibRC + } + + var writer io.Writer = fi + + if sparseHeader > 0 { + writer = sparsefile.NewSparseFileWriter(fi, sparseHeader, totalSize) + } + + bar := pb.Full.Start64(totalSize) + barReader := bar.NewProxyReader(reader) + + _, err = io.Copy(writer, barReader) + return err + } + return fmt.Errorf("no available download url") + }) + + type SparseHoleOptions struct { + FILE string + } + R(&SparseHoleOptions{}, "sparse-file-hole", "Show sparse file holes", func(s *mcclient.ClientSession, args *SparseHoleOptions) error { + fi, err := os.Open(args.FILE) + if err != nil { + return err + } + defer fi.Close() + sp, err := sparsefile.NewSparseFileReader(fi) + if err != nil { + return err + } + holes := sp.GetHoles() + printObject(jsonutils.Marshal(holes)) + return nil + }) + } diff --git a/pkg/hostman/downloader/downloader.go b/pkg/hostman/downloader/downloader.go index 15438b0d1e..92e45d38cd 100644 --- a/pkg/hostman/downloader/downloader.go +++ b/pkg/hostman/downloader/downloader.go @@ -25,6 +25,9 @@ import ( "yunion.io/x/log" "yunion.io/x/pkg/errors" + + "yunion.io/x/onecloud/pkg/util/pb" + "yunion.io/x/onecloud/pkg/util/sparsefile" ) const ( @@ -37,13 +40,14 @@ type SDownloadProvider struct { w http.ResponseWriter rateLimit int compress bool + sparse bool } -func NewDownloadProvider(w http.ResponseWriter, compress bool, rateLimit int) *SDownloadProvider { +func NewDownloadProvider(w http.ResponseWriter, compress, sparse bool, rateLimit int) *SDownloadProvider { if rateLimit <= 0 { rateLimit = DEFAULT_RATE_LIMIT } - return &SDownloadProvider{w, rateLimit, compress} + return &SDownloadProvider{w: w, rateLimit: rateLimit, compress: compress, sparse: sparse} } func (d *SDownloadProvider) Start( @@ -63,7 +67,7 @@ func (d *SDownloadProvider) Start( d.w.Header().Add(k, headers.Get(k)) } - log.Infof("Downloader Start Transfer %s, compress %t", downloadFilePath, d.compress) + log.Infof("Downloader Start Transfer %s, compress %t sparse %t rateLimit: %dMiB/s", downloadFilePath, d.compress, d.sparse, d.rateLimit) spath, err := filepath.EvalSymlinks(downloadFilePath) if err == nil { downloadFilePath = spath @@ -78,16 +82,24 @@ func (d *SDownloadProvider) Start( if err != nil { return errors.Wrapf(err, "fi.Stat") } - d.w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size())) - var ( - end = false - chunk = make([]byte, CHUNK_SIZE) - writer io.Writer = d.w - startTime = time.Now() - sendBytes = 0 - writeChunk []byte - ) + var reader io.Reader + reader = fi + + size := stat.Size() + if d.sparse { + sparse, err := sparsefile.NewSparseFileReader(fi) + if err != nil { + return errors.Wrapf(err, "NewSparseFileReader") + } + size = sparse.Size() + d.w.Header().Set("X-Sparse-Header", fmt.Sprintf("%d", sparse.HeaderSize())) + reader = sparse + } + + d.w.Header().Set("Content-Length", fmt.Sprintf("%d", size)) + + var writer io.Writer = d.w if d.compress { zw, err := zlib.NewWriterLevel(d.w, COMPRESS_LEVEL) @@ -100,40 +112,18 @@ func (d *SDownloadProvider) Start( defer zw.Flush() // it's cool } - for !end { - size, err := fi.Read(chunk) - if err != nil { - if err != io.EOF { - log.Errorln(err) - return err - } else { - end = true - } - } + pb := pb.NewProxyReader(reader, size) + pb.SetRateLimit(d.rateLimit) + pb.SetRefreshRate(time.Second * 10) + pb.SetCallback(func() { + log.Infof("transfer %s rate: %.2f MiB p/s percent: %.2f%%", downloadFilePath, pb.Rate(), pb.Percent()) + }) - writeChunk = chunk[:size] - if size, err = writer.Write(writeChunk); err != nil { - log.Errorln(err) - return err - } else { - sendBytes += size - timeDur := time.Now().Sub(startTime) - exceptDur := float64(sendBytes) / 1000.0 / 1000.0 / float64(d.rateLimit) - if exceptDur > timeDur.Seconds() { - time.Sleep(time.Duration(exceptDur-timeDur.Seconds()) * time.Second) - } - } + _, err = io.Copy(writer, pb) + if err != nil { + return errors.Wrapf(err, "io.Copy") } - // if d.compress { - // zw := writer.(*zlib.Writer) - // zw.Flush() - // } - - sendMb := float64(sendBytes) / 1000.0 / 1000.0 - timeDur := time.Now().Sub(startTime) - log.Infof("Send data: %fMB rate: %fMB/sec", sendMb, sendMb/timeDur.Seconds()) - if onDownloadComplete != nil { onDownloadComplete() } diff --git a/pkg/hostman/downloader/downloadhandler.go b/pkg/hostman/downloader/downloadhandler.go index 268c99b273..1b3924b8b4 100644 --- a/pkg/hostman/downloader/downloadhandler.go +++ b/pkg/hostman/downloader/downloadhandler.go @@ -84,6 +84,10 @@ func isCompress(r *http.Request) bool { return r.Header.Get("X-Compress-Content") == "zlib" } +func isSparse(r *http.Request) bool { + return r.Header.Get("X-Sparse-Content") == "true" +} + func download(ctx context.Context, w http.ResponseWriter, r *http.Request) { var ( params, _, _ = appsrv.FetchEnv(ctx, w, r) @@ -91,11 +95,12 @@ func download(ctx context.Context, w http.ResponseWriter, r *http.Request) { action = params[""] rateLimit = options.HostOptions.BandwidthLimit compress = isCompress(r) + sparse = isSparse(r) ) switch action { case "images": - hand := NewImageCacheDownloadProvider(w, compress, rateLimit, id) + hand := NewImageCacheDownloadProvider(w, compress, sparse, rateLimit, id) if !fileutils2.Exists(hand.downloadFilePath()) { httperrors.NotFoundError(ctx, w, "Image cache %s not found", id) } else { @@ -104,7 +109,7 @@ func download(ctx context.Context, w http.ResponseWriter, r *http.Request) { } } case "servers": - hand := NewGuestDownloadProvider(w, compress, rateLimit, id) + hand := NewGuestDownloadProvider(w, compress, sparse, rateLimit, id) if !fileutils2.Exists(hand.fullPath()) { httperrors.NotFoundError(ctx, w, "Guest %s not found", id) } else { @@ -142,8 +147,9 @@ func diskDownload(ctx context.Context, w http.ResponseWriter, r *http.Request) { hostutils.Response(ctx, w, err) } else { var compress = isCompress(r) + var sparse = isSparse(r) hand := NewImageDownloadProvider(w, - compress, options.HostOptions.BandwidthLimit, disk, "") + compress, sparse, options.HostOptions.BandwidthLimit, disk, "") if err := hand.Start(); err != nil { hostutils.Response(ctx, w, err) } @@ -156,8 +162,9 @@ func diskHead(ctx context.Context, w http.ResponseWriter, r *http.Request) { hostutils.Response(ctx, w, err) } else { var compress = isCompress(r) + var sparse = isSparse(r) hand := NewImageDownloadProvider(w, - compress, options.HostOptions.BandwidthLimit, disk, "") + compress, sparse, options.HostOptions.BandwidthLimit, disk, "") if err := hand.HandlerHead(); err != nil { hostutils.Response(ctx, w, err) } @@ -187,8 +194,9 @@ func snapshotDownload(ctx context.Context, w http.ResponseWriter, r *http.Reques hostutils.Response(ctx, w, err) } else { var compress = isCompress(r) + var sparse = isSparse(r) hand := NewSnapshotDownloadProvider(w, - compress, options.HostOptions.BandwidthLimit, snapshotPath) + compress, sparse, options.HostOptions.BandwidthLimit, snapshotPath) if err := hand.Start(); err != nil { hostutils.Response(ctx, w, err) } @@ -201,8 +209,9 @@ func snapshotHead(ctx context.Context, w http.ResponseWriter, r *http.Request) { hostutils.Response(ctx, w, err) } else { var compress = isCompress(r) + var sparse = isSparse(r) hand := NewSnapshotDownloadProvider(w, - compress, options.HostOptions.BandwidthLimit, snapshotPath) + compress, sparse, options.HostOptions.BandwidthLimit, snapshotPath) if err := hand.HandlerHead(); err != nil { hostutils.Response(ctx, w, err) } @@ -215,7 +224,7 @@ func imageCacheHead(ctx context.Context, w http.ResponseWriter, r *http.Request) rateLimit := options.HostOptions.BandwidthLimit compress := isCompress(r) - hand := NewImageCacheDownloadProvider(w, compress, rateLimit, imageId) + hand := NewImageCacheDownloadProvider(w, compress, false, rateLimit, imageId) if err := hand.HandlerHead(); err != nil { hostutils.Response(ctx, w, err) diff --git a/pkg/hostman/downloader/guest_downloader.go b/pkg/hostman/downloader/guest_downloader.go index 7977886b3d..6be15090bc 100644 --- a/pkg/hostman/downloader/guest_downloader.go +++ b/pkg/hostman/downloader/guest_downloader.go @@ -32,10 +32,10 @@ type SGuestDownloadProvider struct { } func NewGuestDownloadProvider( - w http.ResponseWriter, compress bool, rateLimit int, sid string, + w http.ResponseWriter, compress, sparse bool, rateLimit int, sid string, ) *SGuestDownloadProvider { return &SGuestDownloadProvider{ - SDownloadProvider: NewDownloadProvider(w, compress, rateLimit), + SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit), serverId: sid, } } diff --git a/pkg/hostman/downloader/image_downloader.go b/pkg/hostman/downloader/image_downloader.go index 516761e8b6..880cc447fe 100644 --- a/pkg/hostman/downloader/image_downloader.go +++ b/pkg/hostman/downloader/image_downloader.go @@ -33,9 +33,9 @@ type SImageDownloadProvider struct { compressFormat string } -func NewImageDownloadProvider(w http.ResponseWriter, compress bool, rateLimit int, disk storageman.IDisk, compressFormat string) *SImageDownloadProvider { +func NewImageDownloadProvider(w http.ResponseWriter, compress, sparse bool, rateLimit int, disk storageman.IDisk, compressFormat string) *SImageDownloadProvider { return &SImageDownloadProvider{ - SDownloadProvider: NewDownloadProvider(w, compress, rateLimit), + SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit), disk: disk, compressFormat: compressFormat, } diff --git a/pkg/hostman/downloader/imagecache_downloader.go b/pkg/hostman/downloader/imagecache_downloader.go index 576c1d0719..7f89a15e74 100644 --- a/pkg/hostman/downloader/imagecache_downloader.go +++ b/pkg/hostman/downloader/imagecache_downloader.go @@ -30,10 +30,10 @@ type SImageCacheDownloadProvider struct { } func NewImageCacheDownloadProvider( - w http.ResponseWriter, compress bool, rateLimit int, imageId string, + w http.ResponseWriter, compress, sparse bool, rateLimit int, imageId string, ) *SImageCacheDownloadProvider { return &SImageCacheDownloadProvider{ - SDownloadProvider: NewDownloadProvider(w, compress, rateLimit), + SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit), imageId: imageId, } } diff --git a/pkg/hostman/downloader/snapshot_downloader.go b/pkg/hostman/downloader/snapshot_downloader.go index bb1747b7d0..7da3ed3e64 100644 --- a/pkg/hostman/downloader/snapshot_downloader.go +++ b/pkg/hostman/downloader/snapshot_downloader.go @@ -26,10 +26,10 @@ type SSnapshotDownloadProvider struct { } func NewSnapshotDownloadProvider( - w http.ResponseWriter, compress bool, rateLimit int, snapshotPath string, + w http.ResponseWriter, compress, sparse bool, rateLimit int, snapshotPath string, ) *SSnapshotDownloadProvider { return &SSnapshotDownloadProvider{ - SDownloadProvider: NewDownloadProvider(w, compress, rateLimit), + SDownloadProvider: NewDownloadProvider(w, compress, sparse, rateLimit), snapshotPath: snapshotPath, } } diff --git a/pkg/hostman/storageman/remotefile/remotefile.go b/pkg/hostman/storageman/remotefile/remotefile.go index da98d4d3c8..5e3a18b6ad 100644 --- a/pkg/hostman/storageman/remotefile/remotefile.go +++ b/pkg/hostman/storageman/remotefile/remotefile.go @@ -31,6 +31,8 @@ import ( "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/httputils" + "yunion.io/x/onecloud/pkg/util/pb" + "yunion.io/x/onecloud/pkg/util/sparsefile" ) type SImageDesc struct { @@ -194,6 +196,7 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback if r.compress { header.Set("X-Compress-Content", "zlib") } + header.Set("X-Sparse-Content", "true") if len(r.extraHeaders) > 0 { for k, v := range r.extraHeaders { header.Set(k, v) @@ -214,6 +217,7 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback return errors.Wrapf(err, "request %s %s", method, url) } totalSize, _ := strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64) + sparseHeader, _ := strconv.ParseInt(resp.Header.Get("X-Sparse-Header"), 10, 64) defer resp.Body.Close() if resp.StatusCode < 300 { if getData { @@ -234,38 +238,24 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback defer zlibRC.Close() reader = zlibRC } - var finishChan = make(chan struct{}) - go func() { - defer recover() - preSizeMb := int64(0) - for { - select { - case <-time.After(1 * time.Second): - info, err := fi.Stat() - if err != nil { - log.Errorf("failed stat file %s", r.tmpPath) - return - } - percentInfo, percent := "", 0.0 - if totalSize > 0 { - percent = float64(info.Size()) / float64(totalSize) * 100 - percentInfo = fmt.Sprintf("(%.2f%%)", percent) - } - log.Infof("written file %s size %dM%s", r.tmpPath, info.Size()/1024/1024, percentInfo) - if callback != nil && percent > 0 { - callback(percent, float64(info.Size()-preSizeMb)/1024/1024, totalSize/1024/1024) - } - preSizeMb = info.Size() - case <-finishChan: - if callback != nil { - callback(100, 0, totalSize/1024/1024) - } - return - } + + var writer io.Writer = fi + + if sparseHeader > 0 { + writer = sparsefile.NewSparseFileWriter(fi, sparseHeader, totalSize) + } + + pb := pb.NewProxyReader(reader, totalSize) + pb.SetCallback(func() { + if callback != nil { + go func() { + callback(pb.Percent(), pb.Rate(), totalSize/1024/1024) + }() } - }() - _, err = io.Copy(fi, reader) - close(finishChan) + log.Infof("written file %s rate: %.2f MiB p/s percent: %.2f%%", r.tmpPath, pb.Rate(), pb.Percent()) + }) + + _, err = io.Copy(writer, pb) if err != nil { return errors.Wrapf(err, "io.Copy to tmpPath %s from reader", r.tmpPath) } diff --git a/pkg/util/pb/doc.go b/pkg/util/pb/doc.go new file mode 100644 index 0000000000..ed1bb73ec3 --- /dev/null +++ b/pkg/util/pb/doc.go @@ -0,0 +1 @@ +package pb // import "yunion.io/x/onecloud/pkg/util/pb" diff --git a/pkg/util/pb/pb.go b/pkg/util/pb/pb.go new file mode 100644 index 0000000000..6136bd68f4 --- /dev/null +++ b/pkg/util/pb/pb.go @@ -0,0 +1,134 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package pb + +import ( + "context" + "io" + "time" + + "golang.org/x/time/rate" +) + +const ( + BURSTS = 1024 * 1024 * 1024 +) + +type SProxyReader struct { + reader io.Reader + size int64 + cur int64 + lastCur int64 + + finish chan bool + + debug bool + + ticker *time.Ticker + + start time.Time + callback func() + + ctx context.Context + + refreshRate time.Duration + + // Mbps + rateLimit *rate.Limiter +} + +func NewProxyReader(reader io.Reader, size int64) *SProxyReader { + return &SProxyReader{ + reader: reader, + size: size, + finish: make(chan bool), + ctx: context.Background(), + refreshRate: time.Second * 1, + } +} + +func (self *SProxyReader) SetRateLimit(mb int) { + self.rateLimit = rate.NewLimiter(rate.Limit(mb*1024*1024), BURSTS) + self.rateLimit.AllowN(time.Now(), BURSTS) +} + +// 设置刷新频率 +// 仅在读取数据前生效 +func (self *SProxyReader) SetRefreshRate(rate time.Duration) { + if rate > time.Second && self.cur == 0 { + self.refreshRate = rate + } +} + +func (self *SProxyReader) SetCallback(callback func()) { + self.callback = callback +} + +func (self *SProxyReader) Percent() float64 { + return float64(self.cur) / float64(self.size) * 100.0 +} + +func (self *SProxyReader) AvgRate() float64 { + return float64(self.cur) / float64(1024) / float64(1024) / float64(time.Now().Sub(self.start).Seconds()) +} + +func (self *SProxyReader) Rate() float64 { + return float64(self.cur-self.lastCur) / float64(1024) / float64(1024) / float64(self.refreshRate.Seconds()) +} + +func (self *SProxyReader) Read(p []byte) (n int, err error) { + defer func() { + if err != nil { + self.finish <- true + close(self.finish) + } + }() + + if self.start.IsZero() { + self.start = time.Now() + self.ticker = time.NewTicker(self.refreshRate) + go self.refresh(self.finish) + } + n, err = self.reader.Read(p) + if err != nil { + return n, err + } + if self.rateLimit != nil { + err = self.rateLimit.WaitN(self.ctx, n) + if err != nil { + return n, err + } + } + self.cur += int64(n) + return n, nil +} + +func (self *SProxyReader) refresh(finishChan chan bool) { + defer self.ticker.Stop() + + for { + select { + case <-self.ticker.C: + if self.callback != nil { + self.callback() + } + self.lastCur = self.cur + case finished := <-finishChan: + if finished { + return + } + } + } +} diff --git a/pkg/util/sparsefile/doc.go b/pkg/util/sparsefile/doc.go new file mode 100644 index 0000000000..aebb362d7b --- /dev/null +++ b/pkg/util/sparsefile/doc.go @@ -0,0 +1 @@ +package sparsefile // import "yunion.io/x/onecloud/pkg/util/sparsefile" diff --git a/pkg/util/sparsefile/hole_other.go b/pkg/util/sparsefile/hole_other.go new file mode 100644 index 0000000000..1e5e755d66 --- /dev/null +++ b/pkg/util/sparsefile/hole_other.go @@ -0,0 +1,26 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build !linux +// +build !linux + +package sparsefile + +import ( + "os" +) + +func detectHoles(file *os.File) ([]sSparseHole, error) { + return []sSparseHole{}, nil +} diff --git a/pkg/util/sparsefile/hole_unix.go b/pkg/util/sparsefile/hole_unix.go new file mode 100644 index 0000000000..cf8e9b25d2 --- /dev/null +++ b/pkg/util/sparsefile/hole_unix.go @@ -0,0 +1,60 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build linux +// +build linux + +package sparsefile + +import ( + "io" + "os" + "syscall" + + "golang.org/x/sys/unix" +) + +const ( + SEEK_DATA = 3 + SEEK_HOLE = 4 +) + +func detectHoles(file *os.File) ([]sSparseHole, error) { + holes := []sSparseHole{} + + offset := int64(0) + for { + start, err := unix.Seek(int(file.Fd()), offset, SEEK_HOLE) + if err != nil { + if e, ok := err.(syscall.Errno); ok && e == syscall.ENXIO { + break + } + return nil, err + } + end, err := unix.Seek(int(file.Fd()), start, SEEK_DATA) + if err != nil { + if e, ok := err.(syscall.Errno); ok && e == syscall.ENXIO { + end, _ = file.Seek(0, io.SeekEnd) + if end > start { + holes = append(holes, sSparseHole{Offset: start, Length: end - start}) + } + break + } + return nil, err + } + offset = end + holes = append(holes, sSparseHole{Offset: start, Length: end - start}) + } + return holes, nil +} diff --git a/pkg/util/sparsefile/sparse.go b/pkg/util/sparsefile/sparse.go new file mode 100644 index 0000000000..c79c2af63d --- /dev/null +++ b/pkg/util/sparsefile/sparse.go @@ -0,0 +1,224 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sparsefile + +import ( + "bytes" + "encoding/json" + "io" + "os" + + "yunion.io/x/pkg/errors" +) + +type sSparseHole struct { + Offset int64 + Length int64 +} + +type SparseFileReader struct { + file *os.File + + holes []sSparseHole + + header []byte + size int64 + headerSize int64 + + realReadLen int64 +} + +func (self *SparseFileReader) Close() error { + return self.file.Close() +} + +func (self *SparseFileReader) HeaderSize() int64 { + return self.headerSize +} + +func (self *SparseFileReader) GetHoles() []sSparseHole { + return self.holes +} + +func (self *SparseFileReader) Size() int64 { + holeSize := int64(0) + for _, hole := range self.holes { + holeSize += hole.Length + } + return self.size - holeSize + self.headerSize +} + +func (self *SparseFileReader) Read(p []byte) (int, error) { + if len(self.header) > 0 { + reader := bytes.NewReader(self.header) + n, err := reader.Read(p) + if err != nil { + return n, err + } + self.header = self.header[n:] + return n, nil + } + for _, hole := range self.holes { + if self.realReadLen > hole.Offset { + continue + } + if self.realReadLen < hole.Offset { + body := io.LimitReader(self.file, hole.Offset-self.realReadLen) + n, err := body.Read(p) + if err != nil { + return n, err + } + self.realReadLen += int64(n) + return n, nil + } else if self.realReadLen == hole.Offset { + n, err := self.file.Seek(hole.Length, io.SeekCurrent) + if err != nil { + return int(n), err + } + self.realReadLen += int64(n) + } + } + return self.file.Read(p) +} + +func (self *SparseFileReader) probeHoles() error { + var err error + self.holes, err = detectHoles(self.file) + if err != nil { + return err + } + if len(self.holes) > 0 { + self.header, err = json.Marshal(self.holes) + if err != nil { + return errors.Wrapf(err, "json.Marshal") + } + self.headerSize = int64(len(self.header)) + } + return nil +} + +func NewSparseFileReader(file *os.File) (*SparseFileReader, error) { + ret := &SparseFileReader{file: file, holes: []sSparseHole{}, realReadLen: 0} + stat, err := ret.file.Stat() + if err != nil { + return nil, errors.Wrapf(err, "Stat") + } + ret.size = stat.Size() + err = ret.probeHoles() + if err != nil { + return nil, err + } + _, err = ret.file.Seek(0, io.SeekStart) + return ret, err +} + +type SparseFileWrite struct { + f *os.File + headerSize int64 + + size int64 + + header []byte + holes []sSparseHole + + bodyWriteLen int64 + + readed int +} + +func (self *SparseFileWrite) Close() error { + return self.f.Close() +} + +type zero struct{} + +func (zero) Read(p []byte) (int, error) { + for index := range p { + p[index] = 0 + } + return len(p), nil +} + +func (self *SparseFileWrite) initHeader() error { + err := json.Unmarshal(self.header, &self.holes) + if err != nil { + return errors.Wrapf(err, "unmarshal header") + } + for _, h := range self.holes { + self.size += h.Length + } + return nil +} + +func (self *SparseFileWrite) Write(p []byte) (int, error) { + if len(self.header) < int(self.headerSize) { + n := int(self.headerSize) - len(self.header) + if len(p) >= n { + self.header = append(self.header, p[:n]...) + err := self.initHeader() + if err != nil { + return n, err + } + self.readed = n + } else { + self.header = append(self.header, p...) + return len(p), nil + } + } + + if self.readed == len(p) { + self.readed = 0 + return len(p), nil + } + + for _, hole := range self.holes { + if self.bodyWriteLen > hole.Offset { + continue + } + if self.bodyWriteLen < hole.Offset { + data := p[self.readed:] + if len(p[self.readed:]) > int(hole.Offset-self.bodyWriteLen) { + data = p[self.readed : hole.Offset-self.bodyWriteLen] + } + n, err := self.f.Write(data) + if err != nil { + return n, err + } + self.readed += n + self.bodyWriteLen += int64(n) + if len(p) == self.readed { + self.readed = 0 + return len(p), nil + } + } else if self.bodyWriteLen == hole.Offset { + n, err := self.f.Seek(hole.Length, io.SeekCurrent) + if err != nil { + return int(n), err + } + self.bodyWriteLen += int64(n) + } + } + return self.f.Write(p) +} + +func NewSparseFileWriter(f *os.File, headerSize int64, size int64) *SparseFileWrite { + return &SparseFileWrite{ + f: f, + headerSize: headerSize, + header: []byte{}, + holes: []sSparseHole{}, + size: size, + } +}