停止悬挂同步法

Question

在进口()中有一个XenAPI方法，它是同步的，它通过委托支持取消。

我有以下方法：

private void UploadImage(.., Func<bool> isTaskCancelled)
{
    try
    {
        HTTP_actions.put_import(
            cancellingDelegate: () => isTaskCancelled(),
            ...);
    }
    catch (HTTP.CancelledException exception)
    {
    }
}

在某些情况下，这个方法HTTP_actions.put_import挂起，对isTaskCancelled()没有反应。在这种情况下，整个应用程序也会挂起。

我可以在一个单独的线程中运行这个方法，并在收到取消信号后强制关闭它，但是这个方法并不总是挂起，有时我想优雅地取消这个方法。只有当这个方法真的被绞死的时候，我才想亲手杀死它。

处理这种情况的最佳方法是什么？

Answer 1

为以下文章撰写博客：http://pranayamr.blogspot.in/2017/12/abortcancel-task.html

从过去的2小时开始，我为你尝试了很多解决方案，我想出了下面的解决方案，请试一试。

class Program
{
   //capture request running that , which need to be cancel in case
   // it take more time 
    static Thread threadToCancel = null;
    static async Task<string> DoWork(CancellationToken token)
    {
        var tcs = new TaskCompletionSource<string>();
        //enable this for your use
    //await Task.Factory.StartNew(() =>
    //{
    //    //Capture the thread
    //    threadToCancel = Thread.CurrentThread;
    //    HTTP_actions.put_import(...); 
    //});
    //tcs.SetResult("Completed");
    //return tcs.Task.Result;

    //comment this whole this is just used for testing 
        await Task.Factory.StartNew(() =>
        {
            //Capture the thread
            threadToCancel = Thread.CurrentThread;

            //Simulate work (usually from 3rd party code)
            for (int i = 0; i < 100000; i++)
            {
                Console.WriteLine($"value {i}");
            }

            Console.WriteLine("Task finished!");
        });

        tcs.SetResult("Completed");
        return tcs.Task.Result;
    }


    public static void Main()
    {
        var source = new CancellationTokenSource();
        CancellationToken token = source.Token;
        DoWork(token);
        Task.Factory.StartNew(()=>
        {
            while(true)
            {
                if (token.IsCancellationRequested && threadToCancel!=null)
                {
                    threadToCancel.Abort();
                    Console.WriteLine("Thread aborted");
                }
            }
        });
        ///here 1000 can be replace by miliseconds after which you want to 
        // abort thread which calling your long running method 
        source.CancelAfter(1000);
        Console.ReadLine();   
    }
}

Answer 2

HTTP_actions.put_import调用HTTP_actions.put调用HTTP.put调用HTTP.CopyStream

委托传递给CopyStream，后者检查函数是否为null (未传递)或true (返回值)。但是，它只在While语句中这样做，因此导致阻塞操作的很可能是Stream的Read。不过，如果使用progressDelegate，也可能会出现这种情况。

为了解决这个问题，将对HTTP.put_import()的调用放在任务或后台线程中，然后单独检查是否取消或从任务/线程返回。

有趣的是，快速浏览一下CopyStream代码就发现了一个bug。如果在进程被取消时计算出的函数返回一个基于它正在进行的检查的不同值，那么您实际上可以在不生成CancelledException()的情况下使循环退出。CancelledException调用的结果应该存储在局部变量中。

Answer 3

这是我的最后实现(基于Pranay Rana的回答)。

public class XenImageUploader : IDisposable
{
    public static XenImageUploader Create(Session session, IComponentLogger parentComponentLogger)
    {
        var logger = new ComponentLogger(parentComponentLogger, typeof(XenImageUploader));

        var taskHandler = new XenTaskHandler(
            taskReference: session.RegisterNewTask(UploadTaskName, logger),
            currentSession: session);

        return new XenImageUploader(session, taskHandler, logger);
    }

    private XenImageUploader(Session session, XenTaskHandler xenTaskHandler, IComponentLogger logger)
    {
        _session = session;
        _xenTaskHandler = xenTaskHandler;
        _logger = logger;

        _imageUploadingHasFinishedEvent = new AutoResetEvent(initialState: false);

        _xenApiUploadCancellationReactionTime = new TimeSpan();
    }

    public Maybe<string> Upload(
        string imageFilePath,
        XenStorage destinationStorage,
        ProgressToken progressToken,
        JobCancellationToken cancellationToken)
    {
        _logger.WriteDebug("Image uploading has started.");

        var imageUploadingThread = new Thread(() =>
            UploadImageOfVirtualMachine(
                imageFilePath: imageFilePath,
                storageReference: destinationStorage.GetReference(),
                isTaskCancelled: () => cancellationToken.IsCancellationRequested));

        imageUploadingThread.Start();

        using (new Timer(
            callback: _ => WatchForImageUploadingState(imageUploadingThread, progressToken, cancellationToken),
            state: null,
            dueTime: TimeSpan.Zero,
            period: TaskStatusUpdateTime))
        {
            _imageUploadingHasFinishedEvent.WaitOne(MaxTimeToUploadSvm);
        }

        cancellationToken.PerformCancellationIfRequested();

        return _xenTaskHandler.TaskIsSucceded
            ? new Maybe<string>(((string) _xenTaskHandler.Result).GetOpaqueReferenceFromResult())
            : new Maybe<string>();
    }

    public void Dispose()
    {
        _imageUploadingHasFinishedEvent.Dispose();
    }

    private void UploadImageOfVirtualMachine(string imageFilePath, XenRef<SR> storageReference, Func<bool> isTaskCancelled)
    {
        try
        {
            _logger.WriteDebug("Uploading thread has started.");

            HTTP_actions.put_import(
                progressDelegate: progress => { },
                cancellingDelegate: () => isTaskCancelled(),
                timeout_ms: -1,
                hostname: new Uri(_session.Url).Host,
                proxy: null,
                path: imageFilePath,
                task_id: _xenTaskHandler.TaskReference,
                session_id: _session.uuid,
                restore: false,
                force: false,
                sr_id: storageReference);

            _xenTaskHandler.WaitCompletion();

            _logger.WriteDebug("Uploading thread has finished.");
        }
        catch (HTTP.CancelledException exception)
        {
            _logger.WriteInfo("Image uploading has been cancelled.");
            _logger.WriteInfo(exception.ToDetailedString());
        }

        _imageUploadingHasFinishedEvent.Set();
    }

    private void WatchForImageUploadingState(Thread imageUploadingThread, ProgressToken progressToken, JobCancellationToken cancellationToken)
    {
        progressToken.Progress = _xenTaskHandler.Progress;

        if (!cancellationToken.IsCancellationRequested)
        {
            return;
        }

        _xenApiUploadCancellationReactionTime += TaskStatusUpdateTime;

        if (_xenApiUploadCancellationReactionTime >= TimeForXenApiToReactOnCancel)
        {
            _logger.WriteWarning($"XenApi didn't cancel for {_xenApiUploadCancellationReactionTime}.");

            if (imageUploadingThread.IsAlive)
            {
                try
                {
                    _logger.WriteWarning("Trying to forcefully abort uploading thread.");

                    imageUploadingThread.Abort();
                }
                catch (Exception exception)
                {
                    _logger.WriteError(exception.ToDetailedString());
                }
            }

            _imageUploadingHasFinishedEvent.Set();
        }
    }

    private const string UploadTaskName = "Xen image uploading";

    private static readonly TimeSpan TaskStatusUpdateTime = TimeSpan.FromSeconds(1);
    private static readonly TimeSpan TimeForXenApiToReactOnCancel = TimeSpan.FromSeconds(10);
    private static readonly TimeSpan MaxTimeToUploadSvm = TimeSpan.FromMinutes(20);

    private readonly Session _session;
    private readonly XenTaskHandler _xenTaskHandler;
    private readonly IComponentLogger _logger;

    private readonly AutoResetEvent _imageUploadingHasFinishedEvent;

    private TimeSpan _xenApiUploadCancellationReactionTime;
}