Действия в очереди для простого класса потоков?

Question

Потому что я не могу поймать RaceOnRCWCleanup и потому, что знаю, что я грязно программирую использование COM-объектов из нескольких потоков без дополнительной безопасности. Мне нужно задать вопрос:

Вопрос: Как разрешить другим потокам помещать рабочие элементы/действия в рабочий поток (например, Invoke/BeginInvoke for Control)? В принципе, другой поток должен быть способен сообщить MyThread «Сделай это! Сделай это!» и do no действие вместо MyThread.

Мне также нужно использовать параметры и вернуть значения. Ресурсы (COM-объекты) должны использоваться только одним потоком. Я думал о создании очереди для перечисления действий для выполнения, но я не понимаю, как добавлять параметры и т. Д. Мой первый дизайн выглядит как код ниже.

---

Редактировать: Для меня также важно, чтобы объект нити был простой снаружи. Поэтому я хочу назвать метод вроде myThread.Connect();, не задумываясь о каких-либо делегатах и ​​других вещах. Объект потока должен использовать делегаты или другие методы в фоновом режиме. На данный момент я использую очередь с дополнительным классом, в котором хранятся вызванный метод, параметры, а затем результат. Результаты доставляются через «завершенное событие». Я не думаю, что это хороший способ оставить это так, потому что все вызовы «действия» в потоке асинхронны.

---

public class MyThread 
{ 
    // private 

    private enum MyEnum { Connect, Disconnect } 
    private Queue<MyEnum> queue = new Queue<MyEnum>(); 

    private void Run() 
    { 
     // loops, does work and can regularly check a queue 
    } 

    // public 

    public void Connect(string address, int parameter1, object parameter2) 
    { 
     // triggers connection to a db or somthing similar 
     // internally it queues an action for this thread to perform 
    } 

    public object Disconnect() 
    { 
     // triggers disconnect from a db or something similar 
     // internally it queues an action for this thread to perform 
     // but must wait for result 
    } 

    public void Start() 
    { 
     // starts the thread 
    } 

    public void Stop() 
    { 
     // stops the thread 
    } 
} 

Answer 1

Простым решением было бы создать однопоточных TPL TaskScheduler и очереди заданий к нему. Встроенный ConcurrentExclusiveSchedulerPairможет быть в состоянии гарантировать единственную фиксированную резьбу; вам нужно посмотреть документацию, чтобы подтвердить это. Если нет, то there are others such as QueuedTaskScheduler.

Answer 2

Это сложно строить из себя, но, как правило, делается с делегатами по следующим направлениям:

public class MyThread 
{ 
    private ConcurrentQueue<Action> queue = new ConcurrentQueue<Action>(); 

    private void Run() 
    { 
     Action action; 
     if (queue.TryDequeue(out action)) 
      action(); 
    } 

    public void Connect(Action action) 
    { 
     queue.Enqueue(action); 
    } 
} 

Пример использования параметров с закрытием над переменными:

 var t = new MyThread(); 
     var myVar = 3; 
     t.Connect(delegate() 
     { 
      Console.WriteLine(myVar); 
     }); 

Если у вас нужно вернуть значение, которое становится более сложным :). Ваша очередь должна быть изменена на тип, который может содержать как Action, так и Func. Затем вы можете добавить такой метод.

public void object Connect(Func<object> function) 

Вам также нужно будет какая-нить ждать, как ManualResetEvent так вызывающему Connect() может ждать ответа.

Answer 3

Как сказал @usr, создание однопоточного TaskScheduler было бы лучшим вариантом. ConcurrentExclusiveSchedulerPair.ExclusiveScheduler гарантирует, что он выполняется исключительно, но не в одном потоке для всех задач. Он может использовать любой поток для выполнения Task. Если вам нужен один фиксированный поток, вам нужно реализовать его самостоятельно.

Custom scheduler topic in Msdn показывает, как реализовать LimitedConcurrencyLevelTaskScheduler. Я изменил его, чтобы сделать его «Single Threaded».

internal class SingleThreadedTaskScheduler : TaskScheduler 
{ 
    private readonly LinkedList<Task> tasks = new LinkedList<Task>(); // protected by lock(_tasks) 

    public SingleThreadedTaskScheduler() 
    { 
     RunThread(); 
    } 

    protected sealed override void QueueTask(Task task) 
    { 
     lock (tasks) 
     { 
      tasks.AddLast(task); 
      Monitor.Pulse(tasks); 
     } 
    } 

    private void RunThread() 
    { 
     new Thread(_ => 
     { 
      while (true) 
      { 
       Task item; 
       lock (tasks) 
       { 
        if (tasks.Count == 0) 
        { 
         Monitor.Wait(tasks); 
        } 
        item = tasks.First.Value; 
        tasks.RemoveFirst(); 
       } 
       base.TryExecuteTask(item); 
      } 
     }).Start(); 
    } 

    protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued) 
    { 
     return false; 
    } 

    protected sealed override bool TryDequeue(Task task) 
    { 
     lock (tasks) 
      return tasks.Remove(task); 
    } 

    public sealed override int MaximumConcurrencyLevel 
    { 
     get { return 1; } 
    } 

    protected sealed override IEnumerable<Task> GetScheduledTasks() 
    { 
     bool lockTaken = false; 
     try 
     { 
      Monitor.TryEnter(tasks, ref lockTaken); 
      if (lockTaken) return tasks; 
      else throw new NotSupportedException(); 
     } 
     finally 
     { 
      if (lockTaken) Monitor.Exit(tasks); 
     } 
    } 
} 

Затем используйте его как

private static readonly TaskScheduler singleThreadedScheduler = new SingleThreadedTaskScheduler(); 

private static void Main() 
{ 
    for (int i = 0; i < 10; i++) 
    { 
     int index = i; 
     Task.Factory.StartNew(() => 
     { 
      Thread.Sleep(100);//Simulate some work 
      Console.WriteLine("Task id {0} Executed in Thread Id {1}", index, Thread.CurrentThread.ManagedThreadId); 
     }, CancellationToken.None, TaskCreationOptions.None, singleThreadedScheduler); 
    } 
}