GHZMRNavigatorProXR/Assets/HotUpdate/COSXMLDownloadTask.cs

using System;
using System.Collections.Generic;
using System.Text;
using COSXML.Model.Object;
using COSXML.Utils;
using COSXML.Model;
using System.IO;
using COSXML.Log;
using COSXML.CosException;
using System.Xml.Serialization;
using COSXML.Common;
using System.Threading;

namespace COSXML.Transfer
{
    public sealed class COSXMLDownloadTask : COSXMLTask
    {
        // user params
        private string localDir;
        public string localFileName;
        private long localFileOffset = 0;
        private long rangeStart = 0L;
        private long rangeEnd = -1L;
        private int maxTasks = 5;
        private long sliceSize = 10 * 1024 * 1024;
        private long divisionSize = 20 * 1024 * 1024;
        private bool enableCrc64Check = false;
        private long singleTaskTimeoutMs = 30 * 1000;

        // concurrency control
        private volatile int activeTasks = 0;
        private int maxRetries = 3;
        private Object syncExit = new Object();
        private bool isExit = false;
        private static ReaderWriterLockSlim resumableFileWriteLock = new ReaderWriterLockSlim();
        private Dictionary<int, DownloadSliceStruct> sliceList = new Dictionary<int, DownloadSliceStruct>();

        // internal requests
        private HeadObjectRequest headObjectRequest;
        private GetObjectRequest getObjectRequest;
        private string localFileCrc64;

        // list of ongoing getObjectRequest
        private List<GetObjectRequest> getObjectRequestsList;

        // resumable info
        private string resumableTaskFile = null;
        private DownloadResumableInfo resumableInfo = null;
        private bool resumable = false;
        private HashSet<string> tmpFilePaths = new HashSet<string>();
        private HashSet<int> sliceToRemove = null;

        // global exception
        private COSXML.CosException.CosClientException gClientExp = null;

        public COSXMLDownloadTask(string bucket, string key, string localDir, string localFileName)
            : base(bucket, key)
        {
            if (localDir.EndsWith(System.IO.Path.DirectorySeparatorChar.ToString(), StringComparison.OrdinalIgnoreCase))
            {
                this.localDir = localDir;
            }
            else
            {
                this.localDir = localDir + System.IO.Path.DirectorySeparatorChar;
            }
            this.localFileName = localFileName;
        }

        public COSXMLDownloadTask(GetObjectRequest request)
            : base(request.Bucket, request.Key)
        {
            this.getObjectRequest = request;
            if (request.localDir.EndsWith(System.IO.Path.DirectorySeparatorChar.ToString(), StringComparison.OrdinalIgnoreCase))
            {
                this.localDir = request.localDir;
            }
            else
            {
                this.localDir = request.localDir + System.IO.Path.DirectorySeparatorChar;
            }
            this.localFileName = request.localFileName;
        }

        public void SetRange(long rangeStart, long rangeEnd)
        {
            this.rangeStart = rangeStart;
            this.rangeEnd = rangeEnd;
        }

        public void SetLocalFileOffset(long localFileOffset)
        {
            this.localFileOffset = localFileOffset;
        }

        public void SetResumableDownload(bool resumable)
        {
            this.resumable = resumable;
        }

        public void SetResumableTaskFile(string file)
        {
            this.resumableTaskFile = file;
        }

        public string GetLocalFileCrc64()
        {
            return localFileCrc64;
        }

        public void SetMaxTasks(int maxTasks)
        {
            if (maxTasks <= 0) 
            {
                throw new COSXML.CosException.CosClientException((int) CosClientError.InvalidArgument, "max tasks cannot be negative or zero");
                return;
            }
            this.maxTasks = maxTasks;
        }

        public void SetSliceSize(long sliceSize)
        {
            if (sliceSize <= 0)
            {
                throw new COSXML.CosException.CosClientException((int) CosClientError.InvalidArgument, "slice size cannot be negative or zero");
                return;
            }
            this.sliceSize = sliceSize;
        }

        public void SetDivisionSize(long divisionSize)
        {
            if (divisionSize <= 0)
            {
                throw new COSXML.CosException.CosClientException((int) CosClientError.InvalidArgument, "division size cannot be negative or zero");
                return;
            }
            this.divisionSize = divisionSize;
        }

        public void SetEnableCRC64Check(bool enableCrc64Check)
        {
            this.enableCrc64Check = enableCrc64Check;
        }

        public void SetSingleTaskTimeoutMs(long singleTaskTimeoutMs)
        {
            if (singleTaskTimeoutMs > 0)
            {
                this.singleTaskTimeoutMs = singleTaskTimeoutMs;
            }
        }

        public void SetMaxRetries(int maxRetries)
        {
            if (maxRetries > 0)
            {
                this.maxRetries = maxRetries;
            }
        }

        private void ComputeSliceList(HeadObjectResult result)
        {
            // slice list can be not empty, if use pause&resume, skip it
            if (this.sliceList.Count != 0)
            {
                return;
            }
            long contentLength = result.size;
            rangeEnd = rangeEnd == -1L || (rangeEnd > contentLength) ? contentLength - 1 : rangeEnd;
            if (rangeEnd - rangeStart + 1 < this.divisionSize)
            {
                DownloadSliceStruct slice = new DownloadSliceStruct();
                slice.partNumber = 1;
                slice.sliceStart = rangeStart;
                slice.sliceEnd = rangeEnd;
                this.sliceList.Add(slice.partNumber, slice);
            }
            else
            {
                long sliceCount = ((rangeEnd - rangeStart) / this.sliceSize) + 1;
                for (int i = 0; i < sliceCount; i++)
                {
                    DownloadSliceStruct slice = new DownloadSliceStruct();
                    slice.partNumber = i + 1;
                    slice.sliceStart = rangeStart + i * this.sliceSize;
                    slice.sliceEnd = 
                        (slice.sliceStart + this.sliceSize > rangeEnd)
                        ? rangeEnd : slice.sliceStart + this.sliceSize - 1;
                    this.sliceList.Add(slice.partNumber, slice);
                }
            }
        }

        internal void Download()
        {
            UpdateTaskState(TaskState.Waiting);
            //对象是否存在
            headObjectRequest = new HeadObjectRequest(bucket, key);

            cosXmlServer.HeadObject(headObjectRequest, delegate (CosResult cosResult)
            {
                lock (syncExit)
                {

                    if (isExit)
                    {
                        return;
                    }
                }

                if (UpdateTaskState(TaskState.Running))
                {
                    HeadObjectResult result = cosResult as HeadObjectResult;
                    ComputeSliceList(result);
                    // resolv resumeInfo
                    if (resumable)
                    {
                        if (resumableTaskFile == null)
                        {
                            resumableTaskFile = localDir + localFileName + ".cosresumabletask";
                        }
                        ResumeDownloadInPossible(result, localDir + localFileName);
                    }
                    // concurrent download
                    ConcurrentGetObject(result.crc64ecma);
                }

            },
            
            delegate (CosClientException clientEx, CosServerException serverEx)
            {
                lock (syncExit)
                {

                    if (isExit)
                    {
                        return;
                    }
                }
                if (UpdateTaskState(TaskState.Failed))
                {
                    if (failCallback != null)
                    {
                        failCallback(clientEx, serverEx);
                    }
                }

            });
        }

        // resolve resumable task file, continue in proper position
        private void ResumeDownloadInPossible(HeadObjectResult result, string localFile)
        {
            this.resumableInfo = DownloadResumableInfo.LoadFromResumableFile(resumableTaskFile);
            if (this.resumableInfo != null)
            {
                if ((result.crc64ecma == null || result.crc64ecma == resumableInfo.crc64ecma) && 
                    this.resumableInfo.eTag == result.eTag && 
                    this.resumableInfo.lastModified == result.lastModified && 
                    this.resumableInfo.contentLength == result.size)
                {
                    // load parts downloaded
                    if (this.resumableInfo.slicesDownloaded != null)
                    {
                        // process downloaded parts
                        foreach (DownloadSliceStruct downloadedSlice in resumableInfo.slicesDownloaded)
                        {
                            // remove from current queue
                            DownloadSliceStruct calculatedSlice;
                            bool ret = this.sliceList.TryGetValue(downloadedSlice.partNumber, out calculatedSlice);
                            if (!ret) {
                                // resumable file broken
                                break;
                            }
                            if (calculatedSlice.sliceStart == downloadedSlice.sliceStart
                                && calculatedSlice.sliceEnd == downloadedSlice.sliceEnd)
                            {
                                if (this.sliceToRemove == null)
                                    this.sliceToRemove = new HashSet<int>();
                                this.sliceToRemove.Add(downloadedSlice.partNumber);
                            }
                            // add to merging list
                            string tmpFileName = localDir + "." + localFileName + ".cosresumable." + downloadedSlice.partNumber;
                            this.tmpFilePaths.Add(tmpFileName);
                        }
                    }
                    else
                    {
                        this.resumableInfo.slicesDownloaded = new List<DownloadSliceStruct>();
                    }
                }
            }
            else
            {
                this.resumableInfo = new DownloadResumableInfo();
                this.resumableInfo.contentLength = result.size;
                this.resumableInfo.crc64ecma = result.crc64ecma;
                this.resumableInfo.eTag = result.eTag;
                this.resumableInfo.lastModified = result.lastModified;
                this.resumableInfo.slicesDownloaded = new List<DownloadSliceStruct>();
                resumableInfo.Persist(resumableTaskFile);
            }
        }

        private bool CompareCrc64(string localFile, string crc64ecma)
        {
            Crc64.InitECMA();
            String hash = String.Empty;

            using (FileStream fs = File.Open(localFile, FileMode.Open))
            {
                byte[] buffer = new byte[2048];
                int bytesRead;
                ulong crc = 0;

                while ((bytesRead = fs.Read(buffer, 0, buffer.Length)) > 0) 
                {
                    ulong partCrc = Crc64.Compute(buffer, 0, bytesRead);
                    if (crc == 0) 
                    {
                        crc = partCrc;
                    }
                    else 
                    {
                        crc = Crc64.Combine(crc, partCrc, bytesRead);
                    }
                }

                localFileCrc64 = crc.ToString();
                return localFileCrc64 == crc64ecma;
            }
        }

        // 发起多线程下载
        private void ConcurrentGetObject(string crc64ecma)
        {
            lock (syncExit) 
            {
                if (isExit) 
                {
                    return;
                }
            }
            // 保障目标路径是存在的
            DirectoryInfo dirInfo = new DirectoryInfo(localDir);
            if (!dirInfo.Exists)
            {
                dirInfo.Create();
            }
            // 控制任务数
            AutoResetEvent resetEvent = new AutoResetEvent(false);
            // 记录成功的分片
            if (sliceToRemove == null)
            {
                sliceToRemove = new HashSet<int>();
            }
            // 记录子任务
            if (getObjectRequestsList == null)
            {
                getObjectRequestsList = new List<GetObjectRequest>();
            }
            int retries = 0;
            // 只抛出最后一条服务端回包
            GetObjectResult downloadResult = null;
            while (sliceList.Count != 0 && retries < maxRetries)
            {
                retries += 1;
                foreach (int partNumber in sliceList.Keys)
                {
                    if (sliceToRemove.Contains(partNumber))
                    {
                        continue;
                    }
                    DownloadSliceStruct slice;
                    bool get_state = sliceList.TryGetValue(partNumber, out slice);
                    if (activeTasks >= maxTasks)
                    {
                        resetEvent.WaitOne();
                    }
                    lock (syncExit)
                    {
                        if (isExit)
                        {
                            return;
                        }
                    }
                    string tmpFileName = "." + localFileName + ".cosresumable." + slice.partNumber;
                    /*
                        以下几种场景需要清理临时文件
                        1. 一次下载中, 重试未下载完成的文件
                        2. Pause之后重入下载, 清理Pause前未下载完成的文件
                        3. 断点续传中, 清理未下载完成的单个分块
                    */
                    FileInfo tmpFileInfo = new FileInfo(localDir + tmpFileName);
                    if (tmpFileInfo.Exists 
                        && tmpFileInfo.Length != (slice.sliceEnd - slice.sliceStart + 1) 
                        && localFileOffset != 0)
                    {
                        System.IO.File.Delete(localDir + tmpFileName);
                    }
                    GetObjectRequest subGetObjectRequest = new GetObjectRequest(bucket, key, localDir, tmpFileName);
                    tmpFilePaths.Add(localDir + tmpFileName);
                    subGetObjectRequest.SetRange(slice.sliceStart, slice.sliceEnd);
                    getObjectRequestsList.Add(subGetObjectRequest);
                    // 计算出来只有一个分块, 而且不是Resume或重试剩的一个, 即不走并发下载, 用GetObject的进度回调给客户端
                    if (progressCallback != null && this.sliceList.Count == 1 && sliceToRemove.Count == 0)
                    {
                        subGetObjectRequest.SetCosProgressCallback(delegate(long completed, long total)
                            {
                                progressCallback(completed, total);
                            }
                        );
                    }
                    Interlocked.Increment(ref activeTasks);
                    cosXmlServer.GetObject(subGetObjectRequest,
                        delegate (CosResult result)
                        {
                            Interlocked.Decrement(ref activeTasks);
                            lock (syncExit)
                            {
                                if (isExit)
                                {
                                    return;
                                }
                            }
                            sliceToRemove.Add(partNumber);
                            if (progressCallback != null && this.sliceList.Count > 1)
                            {
                                long completed = sliceToRemove.Count * this.sliceSize;
                                long total = rangeEnd - rangeStart;
                                if (completed > total)
                                    completed = total;
                                progressCallback(completed, total);
                            }
                            downloadResult = result as GetObjectResult;
                            resetEvent.Set();
                            if (resumable)
                            {
                                // flush done parts
                                this.resumableInfo.slicesDownloaded.Add(slice);
                                this.resumableInfo.Persist(resumableTaskFile);
                            }
                        }, 
                        delegate (CosClientException clientEx, CosServerException serverEx)
                        {
                            Interlocked.Decrement(ref activeTasks);
                            lock (syncExit)
                            {
                                if (isExit)
                                {
                                    return;
                                }
                            }
                            // 对服务端返回的4xx, 不重试, 直接抛异常
                            if (serverEx != null && serverEx.statusCode < 500) {
                                throw serverEx;
                                if (failCallback != null)
                                {
                                    failCallback(null, serverEx);
                                }
                                return;
                            }
                            // 对客户端异常, 全部都重试
                            if (clientEx != null) {
                                gClientExp = clientEx;
                            }
                            resetEvent.Set();
                        }
                    );
                }
                long waitTimeMs = 0;
                int lastActiveTasks = activeTasks;
                while (activeTasks != 0)
                {
                    lock (syncExit)
                    {
                        if (isExit)
                        {
                            return;
                        }
                    }
                    Thread.Sleep(100);
                    if (lastActiveTasks == activeTasks) {
                        /*
                            兼容一种子任务既不成功，也不失败，完全hang住的场景
                            在 .net core + 丢包率高的使用场景下会概率性复现
                            当activeTasks一直不变时，累加一个等待的时间
                            超出 singleTaskTimeoutMs 时，全部清理掉进入下一轮重试
                        */
                        waitTimeMs += 100;
                    } else {
                        waitTimeMs = 0;
                        lastActiveTasks = activeTasks;
                    }
                    if (waitTimeMs > singleTaskTimeoutMs) {
                        foreach(GetObjectRequest subGetObjectRequest in getObjectRequestsList) {
                            try {
                                cosXmlServer.Cancel(subGetObjectRequest);
                            } catch (Exception e) {
                                ;
                            }
                        }
                        getObjectRequestsList.Clear();
                        activeTasks = 0;
                        break;
                    }
                }
                // 从下载列表中移除成功分块
                foreach (int partNumber in sliceToRemove)
                {
                    sliceList.Remove(partNumber);
                }
            }
            if (this.sliceList.Count != 0) 
            {
                if (gClientExp != null)
                {
                    throw gClientExp;
                }
                COSXML.CosException.CosClientException clientEx = new COSXML.CosException.CosClientException
                    ((int)CosClientError.InternalError, "max retries " + retries + " excceed, download fail");
                throw clientEx;
                if (UpdateTaskState(TaskState.Failed))
                {
                    if (failCallback != null)
                    {
                        failCallback(clientEx, null);
                    }
                }
                return;
            }
            // 预期每个分块都下载完成了, 开始顺序合并
            FileMode fileMode = FileMode.OpenOrCreate;
            FileInfo localFileInfo = new FileInfo(localDir + localFileName);
            if (localFileInfo.Exists && localFileOffset == 0 && localFileInfo.Length != rangeEnd - rangeStart + 1)
                fileMode = FileMode.Truncate;
            using (var outputStream = File.Open(localDir + localFileName, fileMode))
            {
                outputStream.Seek(localFileOffset, 0);
                // sort
                List<string> tmpFileList = new List<string>(this.tmpFilePaths);
                tmpFileList.Sort(delegate(string x, string y){
                    int partNumber1 = int.Parse(x.Split(new string[]{"cosresumable."}, StringSplitOptions.None)[1]);
                    int partNumber2 = int.Parse(y.Split(new string[]{"cosresumable."}, StringSplitOptions.None)[1]);
                    return partNumber1 - partNumber2;
                });
                foreach (var inputFilePath in tmpFileList)
                {
                    // tmp not exist, clear everything and ask for retry
                    if (!File.Exists(inputFilePath))
                    {
                        // check if download already completed
                        if (File.Exists(localDir + localFileName))
                        {
                            FileInfo fileInfo = new FileInfo(localDir + localFileName);
                            if (fileInfo.Length == rangeEnd - rangeStart + 1)
                            {
                                foreach (var tmpFile in tmpFileList)
                                {
                                    System.IO.File.Delete(tmpFile);
                                }
                                if (resumableTaskFile != null)
                                {
                                    System.IO.File.Delete(resumableTaskFile);
                                }
                                break;
                            }
                        }
                        // not completed, report fatal error
                        foreach (var tmpFile in tmpFileList)
                        {
                            System.IO.File.Delete(tmpFile);
                        }
                        if (resumableTaskFile != null)
                        {
                            System.IO.File.Delete(resumableTaskFile);
                        }
                        if (File.Exists(localDir + localFileName))
                        {
                            System.IO.File.Delete(localDir + localFileName);
                        }
                        COSXML.CosException.CosClientException clientEx = new COSXML.CosException.CosClientException
                            ((int)CosClientError.InternalError, "local tmp file not exist, could be concurrent writing same file"
                            + inputFilePath +" download again");
                        throw clientEx;
                        if (failCallback != null)
                        {
                            failCallback(clientEx, null);
                        }
                        break;
                    }
                    using (var inputStream = File.OpenRead(inputFilePath))
                    {
                        FileInfo info = new FileInfo(inputFilePath);
                        inputStream.CopyTo(outputStream);
                    }
                }
                tmpFileList.Clear();
                tmpFilePaths.Clear();
                outputStream.Close();
                // 合并完成后，默认进行文件大小的检查
                FileInfo completedFileInfo = new FileInfo(localDir + localFileName);
                if (completedFileInfo.Length != rangeEnd - rangeStart + 1) {
                    COSXML.CosException.CosClientException clientEx = new COSXML.CosException.CosClientException
                        ((int)CosClientError.InternalError, "local File Length " + completedFileInfo.Length + 
                        " does not equals to applied download length " + (rangeEnd - rangeStart + 1) + ", try again");
                    throw clientEx;
                    if (UpdateTaskState(TaskState.Failed))
                    {
                        if (failCallback != null)
                        {
                            failCallback(clientEx, null);
                        }
                    }
                    return;
                }
                // 按需进行CRC64的检查
                if (enableCrc64Check) {
                    if (!CompareCrc64(localDir + localFileName, crc64ecma)) {
                        COSXML.CosException.CosClientException clientEx = new COSXML.CosException.CosClientException
                        ((int)CosClientError.CRC64ecmaCheckFailed, "local File Crc64 " + 
                        " does not equals to crc64ecma on cos, try download again");
                        throw clientEx;
                        if (UpdateTaskState(TaskState.Failed))
                        {
                            if (failCallback != null)
                            {
                                failCallback(clientEx, null);
                            }
                        }
                        return;
                    }
                }
                if (UpdateTaskState(TaskState.Completed))
                {
                    var dir = new DirectoryInfo(localDir);
                    foreach (var file in dir.EnumerateFiles("." + localFileName + ".cosresumable.*")) {
                        file.Delete();
                    }
                    if (resumableTaskFile != null)
                    {
                        FileInfo info = new FileInfo(resumableTaskFile);
                        if (info.Exists)
                        {
                            info.Delete();
                        }
                    }
                    //outputStream.Close();
                    DownloadTaskResult downloadTaskResult = new DownloadTaskResult();
                    downloadTaskResult.SetResult(downloadResult);
                    if (successCallback != null)
                    {
                        successCallback(downloadTaskResult);
                    }
                    return;
                } else {
                    // 容灾 return
                    DownloadTaskResult downloadTaskResult = new DownloadTaskResult();
                    downloadTaskResult.SetResult(downloadResult);
                    //outputStream.Close();
                    if (successCallback != null)
                    {
                        successCallback(downloadTaskResult);
                    }
                }
            }
            return;
        }

        private void RealCancle()
        {
            // 停止可能进行中的Head请求
            cosXmlServer.Cancel(headObjectRequest);
            // 停止可能进行中的下载线程
            foreach (GetObjectRequest subGetObjectRequest in getObjectRequestsList) {
                cosXmlServer.Cancel(subGetObjectRequest);
            }
            getObjectRequestsList.Clear();
        }

        private void Clear()
        {
            // delete tmp files
            var dir = new DirectoryInfo(localDir);
            foreach (var file in dir.EnumerateFiles("." + localFileName + ".cosresumable.*")) {
                file.Delete();
            }
            // delete resumable file
            if (resumableTaskFile != null)
            {
                FileInfo info = new FileInfo(resumableTaskFile);
                if (info.Exists)
                {
                    info.Delete();
                }
            }
        }

        public override void Pause()
        {

            if (UpdateTaskState(TaskState.Pause))
            {
                //exit download
                lock (syncExit) 
                {
                    isExit = true; 
                }
                //cancle request
                RealCancle();
            }
        }

        public override void Cancel()
        {

            if (UpdateTaskState(TaskState.Cancel))
            {
                //exit copy
                lock (syncExit) 
                { 
                    isExit = true; 
                }

                //cancle request
                RealCancle();
                //clear recoder
                Clear();
                // throw exception if requested
                if (throwExceptionIfCancelled) {
                    throw new CosClientException((int)CosClientError.UserCancelled, "Download Task Cancelled by user");
                }
            }
        }

        public override void Resume()
        {

            if (UpdateTaskState(TaskState.Resume))
            {
                lock (syncExit)
                {
                    //continue to download
                    isExit = false;
                }
                Download();
            }
        }

        public class DownloadTaskResult : CosResult
        {
            public string eTag;

            public void SetResult(GetObjectResult result)
            {
                this.eTag = result.eTag;
                this.httpCode = result.httpCode;
                this.httpMessage = result.httpMessage;
                this.responseHeaders = result.responseHeaders;
            }

            public override string GetResultInfo()
            {

                return base.GetResultInfo() + ("\n : ETag: " + eTag);
            }
        }

        public sealed class DownloadResumableInfo 
        {
            public string lastModified;

            public long contentLength;

            public string eTag;

            public string crc64ecma;

            public List<DownloadSliceStruct> slicesDownloaded = new List<DownloadSliceStruct>();

            public static DownloadResumableInfo LoadFromResumableFile(string taskFile)
            {
                try
                {
                    using (FileStream stream = File.OpenRead(taskFile))
                    {
                        DownloadResumableInfo resumableInfo = XmlParse.Deserialize<DownloadResumableInfo>(stream);
                        return resumableInfo;
                    }
                }
                catch (System.Exception)
                {
                    return null;
                }
            }

            public void Persist(string taskFile)
            {
                try
                {
                    resumableFileWriteLock.EnterWriteLock();
                    string xml = XmlBuilder.Serialize(this);
                    using (FileStream stream = File.Create(taskFile))
                    {
                        byte[] info = new UTF8Encoding(true).GetBytes(xml);
                        stream.Write(info, 0, info.Length);
                    }
                }
                catch (Exception)
                {
                    return;
                }
                finally
                {
                    resumableFileWriteLock.ExitWriteLock();
                }
            }
        }
    }
}