Помогите UI Dispatcher справиться с потоком вызовов методов

Question

Следующий пост стал немного длиннее, чем ожидалось, я прошу прощения за это, но, возможно, вам будет интересно читать, и, возможно, у вас есть идея, чтобы помочь мне:)

Я занимаюсь разработкой небольшого приложения, графический интерфейс которого состоит из нескольких элементов управления List. Каждый элемент управления List имеет связанный с ним поток, который постоянно создает строки, которые добавляются в список.

Чтобы позволить элементам управления List обновляться различными потоками, я создал расширенную коллекцию ObservableCollection, которая асинхронно вызывает все свои операции для диспетчера пользовательского интерфейса, который работает довольно хорошо. Вот фрагмент кода этого класса для примера операции вставки:

public class ThreadSaveObservableCollection<T> : ObservableCollection<T> {

    private int _index;

    private Dispatcher _uiDispatcher;
    private ReaderWriterLock _rwLock;

    // ...

    private bool _insertRegFlag;

    new public void Insert (int index, T item) {

        if (Thread.CurrentThread == _uiDispatcher.Thread) {

            insert_(index, item);
        } else {

            if (_insertRegFlag) { }
            else {

                BufferedInvoker.RegisterMethod(_index + "." + (int)Methods.Insert);
                _insertRegFlag = true;
            }

            BufferedInvoker.AddInvocation(new Invocation<int, T> { Ident = _index + "." + (int)Methods.Insert, Dispatcher = _uiDispatcher, Priority = DispatcherPriority.Normal, Param1 = index, Param2 = item, Method = new Action<int, T>(insert_) });
        }
    }

    private void insert_ (int index, T item) {

        _rwLock.AcquireWriterLock(Timeout.Infinite);

        DateTime timeStampA = DateTime.Now;

        base.Insert(index, item);

        DateTime timeStampB = DateTime.Now;

        BufferedInvoker.Returned(_index + "." + (int)Methods.Insert, timeStampB.Subtract(timeStampA).TotalMilliseconds);

        _rwLock.ReleaseWriterLock();
    }

    // ...
}

Чтобы смоделировать вызов в форме некой задачи invocation-task, я построил следующее:

public interface IInvocation {

    string Ident { get; set; }
    void Invoke ();
}

public struct Invocation : IInvocation {

    public string Ident { get; set; }
    public Dispatcher Dispatcher { get; set; }
    public DispatcherPriority Priority { get; set; }
    public Delegate Method { get; set; }

    public void Invoke () {

        Dispatcher.BeginInvoke(Method, Priority, new object[] { });
    }
}

Моя проблема сейчас заключается в том, что из-за огромного количества вызовов методов, которые я вызываю в диспетчере пользовательского интерфейса (у меня есть примерно 8-10 потоков, которые постоянно создают строки, которые они добавляют в свои списки), мой пользовательский интерфейс теряет способность отвечать пользователю I / O (например, с помощью мыши) после aprox. 30 секунд, пока он не примет никакого взаимодействия с пользователем через минуту.

Чтобы решить эту проблему, я написал своего рода буферизованный вызов, который отвечает за буферизацию всех вызовов методов, которые я хочу вызвать на диспетчере пользовательского интерфейса, чтобы затем вызывать их контролируемым образом, например, с некоторой задержкой между вызовами, чтобы избежать переполнения диспетчера пользовательского интерфейса.

Вот некоторый код для иллюстрации того, что я делаю (см. Описание после сегмента кода):

public static class BufferedInvoker {

    private static long _invoked;
    private static long _returned;
    private static long _pending;
    private static bool _isInbalanced;

    private static List<IInvocation> _workLoad;
    private static Queue<IInvocation> _queue;

    private static Thread _enqueuingThread;
    private static Thread _dequeuingThread;
    private static ManualResetEvent _terminateSignal;
    private static ManualResetEvent _enqueuSignal;
    private static ManualResetEvent _dequeueSignal;

    public static void AddInvocation (IInvocation invocation) {

        lock (_workLoad) {

            _workLoad.Add(invocation);
            _enqueuSignal.Set();
        }
    }

    private static void _enqueuing () {

        while (!_terminateSignal.WaitOne(0, false)) {

            if (_enqueuSignal.WaitOne()) {

                lock (_workLoad) {

                    lock (_queue) {

                        if (_workLoad.Count == 0 || _queue.Count == 20) {

                            _enqueuSignal.Reset();
                            continue;
                        }

                        IInvocation item = _workLoad[0];
                        _workLoad.RemoveAt(0);
                        _queue.Enqueue(item);

                        if (_queue.Count == 1) _dequeueSignal.Set();
                    }
                }
            }
        }
    }

    private static void _dequeuing () {

        while (!_terminateSignal.WaitOne(0, false)) {

            if (_dequeueSignal.WaitOne()) {

                lock (_queue) {

                    if (_queue.Count == 0) {

                        _dequeueSignal.Reset();
                        continue;
                    }

                    Thread.Sleep(delay);

                    IInvocation i = _queue.Dequeue();
                    i.Invoke();

                    _invoked++;
                    _waiting = _triggered - _invoked;
                }
            }
        }
    }

    public static void Returned (string ident, double duration) {

        _returned++;

        // ...
    }
}

Идея этого BufferedInvoker заключается в том, что ObservableCollections не вызывают операции самостоятельно, а вместо этого вызывают метод AddInvocation объекта BufferedInvoker, который помещает задачу вызова в свой список _workload. Затем BufferedInvoker поддерживает два "внутренних" потока, которые работают с _queue - один поток берет вызовы из списка _workload и помещает их в _queue, а другой поток помещает вызовы из _queue и, наконец, вызывает их один за другим.

Так что это не что иное, как два буфера для хранения отложенных задач вызова, чтобы задержать их фактический вызов. Далее я подсчитываю количество задач вызова, которые фактически были вызваны потоком _dequeuing (то есть long _invoked), и количество методов, которые были возвращены при их выполнении (каждый метод в ObservableCollection вызывает метод Returned() BufferedInvoker, когда он завершает свое выполнение - число, которое хранится в переменной _returned.

Моя идея состояла в том, чтобы получить число ожидающих вызовов с помощью (_invoked - _returned), чтобы получить представление о рабочей нагрузке диспетчера пользовательского интерфейса, но, что удивительно, _pending всегда меньше 1 или 2.

Поэтому моя проблема сейчас в том, что, хотя я откладываю вызовы методов для диспетчера пользовательского интерфейса (с помощью Thread.Sleep(delay)), приложение через некоторое время начинает задерживаться, что отражает тот факт, что пользовательский интерфейс должен выполнить слишком много для обработки пользовательский ввод / вывод.

Но - и это то, что мне действительно интересно - счетчик _pending никогда не достигает высокого значения, в большинстве случаев он равен 0, даже если пользовательский интерфейс уже заморожен.

Так что теперь я должен найти

(1) способ измерить рабочую нагрузку диспетчера пользовательского интерфейса, чтобы определить точку, в которой диспетчер пользовательского интерфейса перегружен, и

(2) сделать что-то против этого.

Так что теперь большое спасибо за чтение до этого момента, и я надеюсь, что у вас есть какие-либо идеи, как вызвать произвольное большое количество методов для диспетчера пользовательского интерфейса, не перегружая его.

Заранее спасибо... выделенный текст * выделенный текст *

Answer 1

Взглянув на меня, я замечаю, что ты спишь с зажатым замком. Это означает, что во время сна никто не может ставить в очередь, делая очередь бесполезной.

Приложение не отстает, потому что очередь занята, а потому что блокировка удерживается почти всегда.

Я думаю, что вам будет лучше удалить все свои реализованные вручную очереди, блокировки и мониторы и просто использовать встроенный ConcurrentQueue. Одна очередь на элемент управления и поток пользовательского интерфейса и один таймер на очередь.

Во всяком случае, вот что я предлагаю:

ConcurrentQueue<Item> queue = new ...;

//timer pulls every 100ms or so
var timer = new Timer(_ => {
 var localItems = new List<Item>();
 while(queue.TryDequeue(...)) { localItems.Add(...); }
 if(localItems.Count != 0) { pushToUI(localItems); }
});

//producer pushes unlimited amounts
new Thread(() => { while(true) queue.Enqueue(...); });

Просто.