Есть некоторый List<SomeClass>, элементы которого должны обработать 4 потока.
Я хотел делать в SomeClass свойство, которое символизировало бы, что некотый поток захватил сей объект на обработку. Другие потоки глядя на такое развитие событий шли бы дальше и брали свободные объекты коллекции.

SomeClass
{
public object m_lock;
private volatile bool attached false;
public bool Attach
{
lock(m_lock)
{
  attached = true;
return attached;
}
}
}

Насколько криво так делать? Как это лучше сделать? Есть еще библиотека FX Parallelis кажется... она вроде будет в 2010 студии... А сейчас ее скачать нельзя, насколько я понимаю?

Здравствуйте, Ellin, Вы писали:

E>Есть некоторый List<SomeClass>, элементы которого должны обработать 4 потока.
E>Я хотел делать в SomeClass свойство, которое символизировало бы, что некотый поток захватил сей объект на обработку. Другие потоки глядя на такое развитие событий шли бы дальше и брали свободные объекты коллекции.

E>SomeClass
E>{
E>public object m_lock;
E>private volatile bool attached false;
E>public bool Attach
E>{
E>lock(m_lock)
E>{
E> attached = true;
E>return attached;
E>}
E>}
E>}
E>Насколько криво так делать? Как это лучше сделать? Есть еще библиотека FX Parallelis кажется... она вроде будет в 2010 студии... А сейчас ее скачать нельзя, насколько я понимаю?

Скачать можно http://en.wikipedia.org/wiki/Parallel_FX_Library , только что там с лицензией не знаю, можно это в продакшене использовать или нет.

Здравствуйте, Ellin, Вы писали:

E>Есть некоторый List<SomeClass>, элементы которого должны обработать 4 потока.
E>Я хотел делать в SomeClass свойство, которое символизировало бы, что некотый поток захватил сей объект на обработку. Другие потоки глядя на такое развитие событий шли бы дальше и брали свободные объекты коллекции.
Почитайте в MSDN про классы Mutex и Monitor, там и примеры соответствующие должны быть.

Здравствуйте, Slider_spb, Вы писали:

S_>Здравствуйте, Ellin, Вы писали:

E>>Есть некоторый List<SomeClass>, элементы которого должны обработать 4 потока.
E>>Я хотел делать в SomeClass свойство, которое символизировало бы, что некотый поток захватил сей объект на обработку. Другие потоки глядя на такое развитие событий шли бы дальше и брали свободные объекты коллекции.
S_>Почитайте в MSDN про классы Mutex и Monitor, там и примеры соответствующие должны быть.
Еще раз читать? Сколько можно? Я ж написал, как с использованием монитора хочу делать... Правильно на твой взгляд?

Здравствуйте, Ellin, Вы писали:

E>Я хотел делать в SomeClass свойство, которое символизировало бы, что некотый поток захватил сей объект на обработку. Другие потоки глядя на такое развитие событий шли бы дальше и брали свободные объекты коллекции.

E>SomeClass
E>{
E>public object m_lock;
E>private volatile bool attached false;
E>public bool Attach
E>{
E>lock(m_lock)
E>{
E> attached = true;
E>return attached;
E>}
E>}
E>}

Не совсем понятно, как с помощью Attach-а можно узнать, что объект уже кем-то обрабатывается/обработался. Метод-то всегда будет возвращать true.

Здравствуйте, Lloyd, Вы писали:

L>Не совсем понятно, как с помощью Attach-а можно узнать, что объект уже кем-то обрабатывается/обработался. Метод-то всегда будет возвращать true.
Да что-то я глупость несусветную сморозил... Хорошо.. Есть коллекция объектов — 100 штук. Есть 4 потока. Как правильно обработать эту коллекцию с использованием этих потоков? Если разбиваю на 4 коллекции по 25 штук — плохо. Первые 25 штук могут влет обработаться, а остальные нет... следовательно первый поток будет бестолку простаивать... Надо как-то забирать из коллекции по мере обработки...

Здравствуйте, Ellin, Вы писали:

E>Здравствуйте, Lloyd, Вы писали:

L>>Не совсем понятно, как с помощью Attach-а можно узнать, что объект уже кем-то обрабатывается/обработался. Метод-то всегда будет возвращать true.
E>Да что-то я глупость несусветную сморозил... Хорошо.. Есть коллекция объектов — 100 штук. Есть 4 потока. Как правильно обработать эту коллекцию с использованием этих потоков? Если разбиваю на 4 коллекции по 25 штук — плохо. Первые 25 штук могут влет обработаться, а остальные нет... следовательно первый поток будет бестолку простаивать... Надо как-то забирать из коллекции по мере обработки...

Взять семафор на 4 позиции и кидать в ThreadPool если в семафоре есть свободные места...

Здравствуйте, Lloyd, Вы писали:

L>Не совсем понятно, как с помощью Attach-а можно узнать, что объект уже кем-то обрабатывается/обработался. Метод-то всегда будет возвращать true.
Я хотел как-то так:

    internal class  SomeClass
    {
        private int val = 0;  
        public SomeClass(int val)
        {
            this.val = val;
        }

        private object mlock = new object();
        private volatile bool isProcessed = false;

        public void Process(int i)
        {
            if(Monitor.TryEnter(mlock) == true)
            {
                try
                {
                    if(isProcessed == false)
                    {
                        Console.WriteLine(Thread.CurrentThread.Name+" обрабатывает " + val);
                        val = i;
                        isProcessed = true;
                    }
                }
                finally
                {
                    Monitor.Exit(mlock);                    
                }
            }
        }
        public override string ToString()
        {
            return val.ToString();
        }
    }
    class Program
    {
        static void Main(string[] args)
        {
            List<SomeClass> cls = new List<SomeClass>();                        
            for (int i = 0; i < 100; i++ )
            {                
                cls.Add(new SomeClass(i));
            }
            Thread[] threads =  new Thread[4];
            for (int i = 0; i < 4; i++)
            {
                threads[i] = new Thread(() =>
                                            {
                                                foreach (SomeClass cl in cls)
                                                {
                                                    cl.Process(-1);
                                                }
                                            });
                threads[i].Name = "Поток №" + i;
            }
            foreach (var thread in threads)
            {
                thread.Start();
            }
        }
    }

Однако не всегда срабатывает... иногда в концовке не проходит. Почему интересно?

E>Однако не всегда срабатывает... иногда в концовке не проходит. Почему интересно?
Все ясно... когда главный поток завершается не дает доработаь обработчикам... Ну и все равно данный подход какой-то неоптимальный... т.е. если в коллекции 10000 объектов для "запаздавших" потоков будет слишком много проверок заведомого ложно if(isProcessed == false). Плюс как бы обойтись без volatile... ведь оно сбрасывает кеш процессора, а это плохо сказывается на производительности.

E>>Насколько криво так делать? Как это лучше сделать? Есть еще библиотека FX Parallelis кажется... она вроде будет в 2010 студии... А сейчас ее скачать нельзя, насколько я понимаю?

_>Скачать можно http://en.wikipedia.org/wiki/Parallel_FX_Library , только что там с лицензией не знаю, можно это в продакшене использовать или нет.


Поддерживаю. Благодаря либе можно код свести к такому виду:

// Библиотека запустит не более 4-х потоков. Больше информации в хелпе Parallel Extentions.
                    var taskManager = new TaskManager(new TaskManagerPolicy(4, 4, 1)); // 4 - это Ваше кол-во потоков.
                    Parallel.ForEach(someClasses, someClass =>
                    {
                        DoSomething(someClass); // ваша обработка
                    }, taskManager);

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

Здравствуйте, Ellin, Вы писали:

E>Да что-то я глупость несусветную сморозил... Хорошо.. Есть коллекция объектов — 100 штук. Есть 4 потока. Как правильно обработать эту коллекцию с использованием этих потоков? Если разбиваю на 4 коллекции по 25 штук — плохо. Первые 25 штук могут влет обработаться, а остальные нет... следовательно первый поток будет бестолку простаивать... Надо как-то забирать из коллекции по мере обработки...

стек\фифо чем не подходят?

KS>Поддерживаю. Благодаря либе можно код свести к такому виду:
KS>

KS>// Библиотека запустит не более 4-х потоков. Больше информации в хелпе Parallel Extentions.
KS>                    var taskManager = new TaskManager(new TaskManagerPolicy(4, 4, 1)); // 4 - это Ваше кол-во потоков.
KS>                    Parallel.ForEach(someClasses, someClass =>
KS>                    {
KS>                        DoSomething(someClass); // ваша обработка
KS>                    }, taskManager);
KS>

KS>Причём SomeClass может быть чем угодно, и не придётся городить в нём всякие защиты от одновременного доступа и пр.

Да, я помню что там как-то так делалось... вот только в свое время я эту библиотеку так скачать и не смог (ссылка была закрыта). А сейчас вообще дажу ту ссылку найти не могу. Или под 2008 студией ее так и не получится запустить?

Так почему-то получается слишком медлено... в 4 раза медленнее чем 1 поток

Здравствуйте, Ellin, Вы писали:

E>Да, я помню что там как-то так делалось... вот только в свое время я эту библиотеку так скачать и не смог (ссылка была закрыта). А сейчас вообще дажу ту ссылку найти не могу. Или под 2008 студией ее так и не получится запустить?

Блог — http://blogs.msdn.com/pfxteam/
Форум — http://social.msdn.microsoft.com/Forums/en-US/parallelextensions/threads/
Скачать либу и хелп к ней — http://www.microsoft.com/downloads/details.aspx?FamilyId=348F73FD-593D-4B3C-B055-694C50D2B0F3&amp;displaylang=en

Здравствуйте, Ellin, Вы писали:

E>Так почему-то получается слишком медлено... в 4 раза медленнее чем 1 поток

так ясен пень... 4 потока постоянно конкурируют за один и тот же объект, и 3 из 4 ждут пока его обработает 1.
попробуйте синхронизировать не обработку, а доступ к списку задач, скажем так:

    internal class  SomeClass
    {
        private int val = 0;  
        public SomeClass(int val)
        {
            this.val = val;
        }

        public void Process(int i)
        {
                    if(isProcessed == false)
                    {
                        Console.WriteLine(Thread.CurrentThread.Name+" обрабатывает " + val);
                        val = i;
                        isProcessed = true;
                    }
        }
        public override string ToString()
        {
            return val.ToString();
        }
    }
    class Program
    {
        static void Main(string[] args)
        {
            Stack cls = new Stack();                        
            for (int i = 0; i < 100; i++ )
            {                
                cls.Push(new SomeClass(i));
            }
            Thread[] threads =  new Thread[4];
            for (int i = 0; i < 4; i++)
            {
                threads[i] = new Thread(() =>
                                            {
                                                bool doNext = true;
                                                while (doNext)
                                                { 
                                                     try
                                                     { 
                                                         SomeClass po;
                                                         lock(cls.SyncRoot)
                                                         {
                                                             po = (SomeClass)cls.Pop() // IOE when Stack is empty
                                                         }

                                                         po.Process(-1);
                                                      }
                                                      catch(InvalidOperationException)
                                                      {
                                                          doNext = false;
                                                      }
                                                 }
                                            });
                threads[i].Name = "Поток №" + i;
            }
            foreach (var thread in threads)
            {
                thread.Start();
            }
        }
    }

Здравствуйте, Ellin, Вы писали:

Не будучи специалистом в C#, я все же отметил бы, что я бы просто хранил позицию в списке вместо того, чтобы заводить в объекте какие-то поля. Есть список, некий поток хочет взять элемент на обработку, он его берет в текущей позиции списка и продвигает позицию, вот и все. Разумееется, делать это надо под монитором, дабы два потока не взяли один и тот же элемент.

Здравствуйте, Ellin, Вы писали:

E>Так почему-то получается слишком медлено... в 4 раза медленнее чем 1 поток

Если на Вашей машине имеется только один процессор(ядро), то даже при отсутствии каких-либо блокировок, 4 потока никак не смогут выполнять приведённый Вами код быстрее, чем это сделает один поток. У Вас 4 потока из каких соображений? Может стоит воспользоваться ThreadPool? Что-то вроде такого:

using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace TestThreadPool
{
    class SomeItem
    {
        public static int CountToProcess = 0;
        public static EventWaitHandle Event = new EventWaitHandle(false, EventResetMode.ManualReset);

        private int _Value;
        public SomeItem(int Val) 
        {
            _Value = Val; 
        }

        public int ProcessContext;
        public void Process(object State)
        {
            SomeItem Item = (SomeItem)State;
            Item._Value += Item.ProcessContext;
//            Thread.Sleep(100);
            if (0 == Interlocked.Decrement(ref CountToProcess))
            {
                Event.Set();
            }
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMaxThreads(Environment.ProcessorCount, Environment.ProcessorCount);

            List<SomeItem> Collection = new List<SomeItem>();
            for (int i = 0; i < 100; i++)
            {
                Collection.Add(new SomeItem(i + 1));
            }

            int V = 1000;
            SomeItem.CountToProcess = Collection.Count;
            SomeItem.Event.Reset();
            foreach (SomeItem Item in Collection)
            {
                Item.ProcessContext = ++V;
                ThreadPool.QueueUserWorkItem(new WaitCallback(Item.Process), Item); 
            }

            Console.WriteLine("Start !");
            if (SomeItem.Event.WaitOne(3000, true)) 
            {
                SomeItem.Event.Reset();
                Console.WriteLine("Complete !"); 
            }
            else 
            {
                Console.WriteLine("Timeout !");
            }
            Console.ReadLine();
        }
    }
}

Здравствуйте, ili, Вы писали:

Так нет, там tryEnter ведь... т.е. если он захвачен, то поток чешет дальше по списку пока свободный не найдет. ИМХО медленно т.к. еще меряется время создание потоков в ОС...

Ну... Intel(R) Core(TM) 2 Duo CPU E6550 @ 2.33 GHz
Я так думаю 4-ре, ну как минимум 2 ядра — столько же и потоков может чесать? Или я не прав?

Здравствуйте, Ellin, Вы писали:

E>Ну... Intel(R) Core(TM) 2 Duo CPU E6550 @ 2.33 GHz
E>Я так думаю 4-ре, ну как минимум 2 ядра — столько же и потоков может чесать? Или я не прав?

Если Ваши потоки во время работы не будут "засыпать" на каких-нибудь операциях ввода-вывода (в широком смысле этого слова ), то есть всё время будут потреблять ресурсы процессоров, то оптимально выбрать число потоков по числу ядер. Но не стоит "зашивать" это число жёстко в код, лучше на месте спросить у системы — GetSystemInfo в WinAPI или Environment.ProcessorCount в NET. Ещё лучше поручить этот выбор системе, воспользовавшись пулом.