Уважаемые, встал в тупик и не знаю что делать, подскажите плз:

Есть задача: многопоточная обработка данных. В частности: необходимо многопоточно анализировать веб страницы. С этих страниц мы выдираем ссылки и их анализируем и т.д.

Как мне это видится:
1. Есть поток, который занимается только скачиванием страницы в нужное место для последующего поиска. По завершению потока выполняется некая функция OnFinishThread,
2. Есть некая функция PushPage(address), которая добавляет страницу в общий список, присваивает странице address состояние типа NEED_DOWNLOAD и все, больше она ничего не делает.
3. Есть некая функция Update, которая просматривает список страниц и для каждой страницы со статусом NEED_DOWNLOAD запускает тот самый поток
4. Функция OnFinishThread берет ту страницу, которую скачал соотв. вызывающий поток, и вызывает функцию ParsePage, которая на каждый найденный урл вызовет соотв. PushAddress и т.д.
Соотв. функция Update циклическая и все скачивается, анализируется и т.д.

Вопрос: с точки зрения многопоточности критичная для нас функция OnFinishThread, т.к. ее могут вызвать сразу несколько потоков одновременно, прально? Соотв. делаем, например, так:

void OnFinishThread(Tthread *Thread)
{
EnterCriticalSection(&CS);
try
{
if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");
++Locker;
try
{
ParsePage(Thread->Page);
}__finally
{
--Locker;
};
}__finally
{
LeaveCriticalSection(&CS);
};
};

Подобный вариант НЕ РАБОТАЕТ ((

Как подобное лучше сделать?!

Hello HAS, you wrote:

> Вопрос: с точки зрения многопоточности критичная для нас функция OnFinishThread, т.к. ее могут вызвать сразу несколько потоков одновременно, прально? Соотв. делаем, например, так:
>
> void OnFinishThread(Tthread *Thread)
> {
> EnterCriticalSection(&CS);
> try
> {
> if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");

Данная проверка бессмысленна, т.к. EnterCriticalSection ГАРАНТИРУЕТ, что данный блок выполняется только одним потоком.
Что не работает?

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

--
Всего хорошего, Слава
ICQ: 197577902

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

SA>Hello HAS, you wrote:

>> Вопрос: с точки зрения многопоточности критичная для нас функция OnFinishThread, т.к. ее могут вызвать сразу несколько потоков одновременно, прально? Соотв. делаем, например, так:
>>
>> void OnFinishThread(Tthread *Thread)
>> {
>> EnterCriticalSection(&CS);
>> try
>> {
>> if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");

SA>Данная проверка бессмысленна, т.к. EnterCriticalSection ГАРАНТИРУЕТ, что данный блок выполняется только одним потоком.

тогда почему срабатывает данное исключение?

SA>Что не работает?
вроде как работает, но: один раз может отработать без задоринки, в другой раз может просто выдать "out of memory", в третий — подвиснуть... причем пытался уже статически задавать последовательность страниц, та же история...

SA>Кроме того, критическую секцию нужно ставить не на всю функцию, а только на те места где идет обращение в разделяемым ресурсам, т.е. ресурсам к котором возможен доступ из нескольких потоков.

а как быть если мне необходимо, чтобы весь этот блок кода был выполнен без прерывания одним потоком, т.е. чтобы в этом блоке в ходе выполнения не было переключения между потоками?

SA>--
SA>Всего хорошего, Слава
SA>ICQ: 197577902

Hello HAS, you wrote:

>> Данная проверка бессмысленна, т.к. EnterCriticalSection ГАРАНТИРУЕТ, что данный блок выполняется только одним потоком.
> тогда почему срабатывает данное исключение?

Какое исключение?

>> Что не работает?
> вроде как работает, но: один раз может отработать без задоринки, в другой раз может просто выдать "out of memory", в третий — подвиснуть... причем пытался уже статически задавать последовательность страниц, та же история...

Это нужно смотреть что у вас в ParsePage(Thread->Page);

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

Зачем? Какой вам тогда вообще смысл в многопоточности если вы ее накорню прибиваете.

--
Всего хорошего, Слава
ICQ: 197577902

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

SA>Hello HAS, you wrote:

>>> Данная проверка бессмысленна, т.к. EnterCriticalSection ГАРАНТИРУЕТ, что данный блок выполняется только одним потоком.
>> тогда почему срабатывает данное исключение?

SA>Какое исключение?

>>> Что не работает?
>> вроде как работает, но: один раз может отработать без задоринки, в другой раз может просто выдать "out of memory", в третий — подвиснуть... причем пытался уже статически задавать последовательность страниц, та же история...

SA>Это нужно смотреть что у вас в ParsePage(Thread->Page);

на данный момент там PushPage(Addresses[random(AddressCounts)]); т.е. просто добавление страницы в очередь

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

SA>Зачем? Какой вам тогда вообще смысл в многопоточности если вы ее накорню прибиваете.

почему? каждый поток по идее может занимать в работе длительное время, т.е. грубо аналог: Teleport Pro — какие-то страницы могут быть недоступны и поток будет ждать таймаута, в то вермя как другой поток в это время скачает другую страницу...
Чем же я прибиваю многопоточность?

SA>--
SA>Всего хорошего, Слава
SA>ICQ: 197577902

<skipped>

HAS>Подобный вариант НЕ РАБОТАЕТ ((

HAS>Как подобное лучше сделать?!

Ты ее инициализируешь? Покажи весь код
См. MSDN::InitializeCriticalSection — может выбрасывать exception

З.Ы. Оберни нативную CS врапперами, замучаешься везде писать try/catch

On Sun, 05 Mar 2006 19:35:26 +0600, HAS <36274@users.rsdn.ru> wrote:

>> Это нужно смотреть что у вас в ParsePage(Thread->Page);
> на данный момент там PushPage(Addresses[random(AddressCounts)]); т.е. просто добавление страницы в очередь

Хм... кстати, может менеджер памяти не знает про многопоточность и начинают лезть глюки с памятью?

По поводу зависаний: если зависло и при этом не есть процессорное время, то дело явно в deadlock.

В любом случае это нужно все прогнать под отладчиком.

>> Зачем? Какой вам тогда вообще смысл в многопоточности если вы ее накорню прибиваете.
> почему?

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

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

--
Всего хорошего, Слава.
ICQ: 197577902

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

HAS>void OnFinishThread(Tthread *Thread)
HAS>{
HAS> EnterCriticalSection(&CS);
HAS> try
HAS> {
HAS> if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");
HAS> ++Locker;
HAS> try
HAS> {
HAS> ParsePage(Thread->Page);
HAS> }__finally
HAS> {
HAS> --Locker;
HAS> };
HAS> }__finally
HAS> {
HAS> LeaveCriticalSection(&CS);
HAS> };
HAS>};

HAS>Подобный вариант НЕ РАБОТАЕТ ((

HAS>Как подобное лучше сделать?!


Здесь вообще не нужна синхронизация, т.к. нет работы с разделяемыми ресурсами. Код можно переписать просто как:

void OnFinishThread(Tthread *Thread)
{
  ParsePage(Thread->Page); 
}

Пока поток работает только с локальными данными и не обращается ни к каким глобальным данным (к которым одновременно обращаются сразу несколько потоков) никакая синхронизация не нужна

On Mon, 06 Mar 2006 09:10:15 +0600, remark <38267@users.rsdn.ru> wrote:

> Здесь вообще не нужна синхронизация, т.к. нет работы с разделяемыми ресурсами.

А как же параметр Thread?
Так что синхронизация нужна.

--
Всего хорошего, Слава.
ICQ: 197577902

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

_>On Mon, 06 Mar 2006 09:10:15 +0600, remark <38267@users.rsdn.ru> wrote:

>> Здесь вообще не нужна синхронизация, т.к. нет работы с разделяемыми ресурсами.

_>А как же параметр Thread?
_>Так что синхронизация нужна.

А кто сказал, что Thread — разделяемый ресурс? Во всяком случае нужно смотреть реализацию из которой это будет видно. А вообще судя по описаной логике, разделяемый ресурс — это очередь куда складываются и откуда достаются готовые, скачанные страницы. Вот обращения к ней push и pop нужно синхронизировать. В остальном синхронизация не нужна, пусть себе работают потоки (каждый со своим ресурсом) и сигналят о готовности.

Вообще я себе представляю задачу так:
1. Стартуем поток, которому передаем параметр — какую страницу качать.
2. По завершении скачивания поток синхронно закидывает готовую страницу в очередь, сигналит о своем завершении=появлении новой страницы в очереди и завершается (как вариант не завершается, а берет из очереди заданий новую страницу которую нужно качать)
3. Поток-монитор получив сигнал, синхронно забирает из очереди готовую страницу и дальше делает с ней что угодно, больше кроме него эту страницу никто не трогает.

Вот на мой взгляд те две точки (выделенные жирным цветом ), которые нужно синхронизировать.

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

HAS>Уважаемые, встал в тупик и не знаю что делать, подскажите плз:

HAS>Вопрос: с точки зрения многопоточности критичная для нас функция OnFinishThread, т.к. ее могут вызвать сразу несколько потоков одновременно, прально? Соотв. делаем, например, так:

HAS>void OnFinishThread(Tthread *Thread)
HAS>{
HAS> EnterCriticalSection(&CS);
HAS> try
HAS> {
HAS> if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");
HAS> ++Locker;
HAS> try
HAS> {
HAS> ParsePage(Thread->Page);
HAS> }__finally
HAS> {
HAS> --Locker;
HAS> };
HAS> }__finally
HAS> {
HAS> LeaveCriticalSection(&CS);
HAS> };
HAS>};

HAS>Подобный вариант НЕ РАБОТАЕТ ((

HAS>Как подобное лучше сделать?!

Ведь при входе в OnFinishThread можно выставить семафор, а перед вызовом OnFinishThread его проверять и оценивать необходимость запуска.

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

_>On Sun, 05 Mar 2006 19:35:26 +0600, HAS <36274@users.rsdn.ru> wrote:

>>> Это нужно смотреть что у вас в ParsePage(Thread->Page);
>> на данный момент там PushPage(Addresses[random(AddressCounts)]); т.е. просто добавление страницы в очередь

_>Хм... кстати, может менеджер памяти не знает про многопоточность и начинают лезть глюки с памятью?

по-идее должен знать... пишу всо под Builder 5.0... кстати, очень часто проект подвисает с ошибкой типа "что-то там сломалось в Borlandbk50.bpk". За что отвесает этот пакет?

и еще: какие моменты не может уловить CodeGuard? потому как с точки зрения кода — везде все очищается как надо, но если забить большой цикл по созданию и убиванию
этого менеджера (порядка > 1000) то память жрется неплохо, а CodeGuard молчит ((


_>По поводу зависаний: если зависло и при этом не есть процессорное время, то дело явно в deadlock.

_>В любом случае это нужно все прогнать под отладчиком.

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

>>> Зачем? Какой вам тогда вообще смысл в многопоточности если вы ее накорню прибиваете.
>> почему?

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

поток только качает страницу и скидывает её в файлик

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

пример работы с этой очередью:

void __fastcall THttpDownloadManager::Update()
{
EnterCriticalSection(&CS);
try
{
#define MaxThreadCount 20
while ((FAddressWaiting > 0 || FAddressDownloaded > 0) && FAddressDownloading < MaxThreadCount && !FPause && !FTerminate)
{
TAddressInfo *Info = NULL;
for (int i=FAddresses->Count-1; i>=0 && FAddressDownloading < MaxThreadCount; --i)
{
Info = (TAddressInfo*)FAddresses->Items[i];
if (Info == NULL) continue;
switch (Info->State)
{
case AS_INORDER: {
if ((FAddressWaiting > 0) && (FAddressDownloading < MaxThreadCount)) DownloadPage(Info);
break;
};
case AS_ERROR: {
++FAddressError;
--FAddressDownloaded;
++FAddressReady;
Info->State = AS_IDLE;
DoAddressStateChange(Info);
break;
};
case AS_DOWNLOADED: {
--FAddressDownloaded;
++FAddressReady;
if (Info->Recursive)
ParsePage(Info);
Info->State = AS_READY;
DoAddressStateChange(Info);
break;
};
};
};
};
if (FAddressReady == FAddressCount | FTerminate)
{
FState = (FTerminate) ? DM_TERMINATE: DM_READY;
if (FOnFinished != NULL) FOnFinished(NULL);
};

}__finally
{
LeaveCriticalSection(&CS);
};
};

---
при этом DownloadPage создает и запускает поток, ParsePage — выдирает из страницы ссылки и добавляет их в очередь через PushPage (pushpage заключен между EnterCriticalSection & LeaveCriticalSection). Чтоздесь не так?

_>--
_>Всего хорошего, Слава.
_>ICQ: 197577902

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

L><skipped>

HAS>>Подобный вариант НЕ РАБОТАЕТ ((

HAS>>Как подобное лучше сделать?!

L>Ты ее инициализируешь? Покажи весь код

инициализация происходит в конструкторе менеджера, а убиение соотв. в деструкторе...

L>См. MSDN::InitializeCriticalSection — может выбрасывать exception

ну тогда наверное и менеджер не стал бы работать? а он работает... вылетает просто от балды (

L>З.Ы. Оберни нативную CS врапперами, замучаешься везде писать try/catch

типа чтобы юзать так:

void ....
{
TCSWrapper Wrapper(&CS); // в конструкторе EnterCriticalSection
...
// а здесь типа срабатывает деструктор с LeaveCriticalSection
};

??? Есть гарантия 100% что деструктор всегда будет отрабатывать во время и верно?

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

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

HAS>>void OnFinishThread(Tthread *Thread)
HAS>>{
HAS>> EnterCriticalSection(&CS);
HAS>> try
HAS>> {
HAS>> if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");
HAS>> ++Locker;
HAS>> try
HAS>> {
HAS>> ParsePage(Thread->Page);
HAS>> }__finally
HAS>> {
HAS>> --Locker;
HAS>> };
HAS>> }__finally
HAS>> {
HAS>> LeaveCriticalSection(&CS);
HAS>> };
HAS>>};

HAS>>Подобный вариант НЕ РАБОТАЕТ ((

HAS>>Как подобное лучше сделать?!


R>Здесь вообще не нужна синхронизация, т.к. нет работы с разделяемыми ресурсами. Код можно переписать просто как:

R>

R>void OnFinishThread(Tthread *Thread)
R>{
R>  ParsePage(Thread->Page); 
R>}
R>

R>Пока поток работает только с локальными данными и не обращается ни к каким глобальным данным (к которым одновременно обращаются сразу несколько потоков) никакая синхронизация не нужна

так Thread->Page и есть глобальные данные, с которыми происходит работа в нескольких местах программы

R>

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

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

_>>On Mon, 06 Mar 2006 09:10:15 +0600, remark <38267@users.rsdn.ru> wrote:

>>> Здесь вообще не нужна синхронизация, т.к. нет работы с разделяемыми ресурсами.

_>>А как же параметр Thread?
_>>Так что синхронизация нужна.

AF>А кто сказал, что Thread — разделяемый ресурс? Во всяком случае нужно смотреть реализацию из которой это будет видно. А вообще судя по описаной логике, разделяемый ресурс — это очередь куда складываются и откуда достаются готовые, скачанные страницы. Вот обращения к ней push и pop нужно синхронизировать. В остальном синхронизация не нужна, пусть себе работают потоки (каждый со своим ресурсом) и сигналят о готовности.

так и есть... при этом как такого pop'а нету, есть Update, который пробегает по всей очереди и в зависимости от статуса страницы либо вызывает ParsePage, либо PushPage, либо создает новый поток

AF>Вообще я себе представляю задачу так:
AF>1. Стартуем поток, которому передаем параметр — какую страницу качать.
AF>2. По завершении скачивания поток синхронно закидывает готовую страницу в очередь, сигналит о своем завершении=появлении новой страницы в очереди и завершается (как вариант не завершается, а берет из очереди заданий новую страницу которую нужно качать)

что значит "синхронно закидывает страницу в очередь" ? т.е. в рамках блока EnterCriticalSection & LeaveeCriticalSection ? или как? и как лучше отсигналить?

AF>3. Поток-монитор получив сигнал, синхронно забирает из очереди готовую страницу и дальше делает с ней что угодно, больше кроме него эту страницу никто не трогает.

поток монитор получается основной поток... опять же что значит синхронно?

AF>Вот на мой взгляд те две точки (выделенные жирным цветом ), которые нужно синхронизировать.

в принципе так все и сделано:
1. есть update, который бегает по очереди и в зависимости от статуса страницы либо разбирает её, либо запускает новый поток
2. есть OnFinishthread, который вызывается по завершению любого потока, скачавшего страницу. он меняет статус страницы на скачанный (в рамках EnterCS & LeaveSC) и вызывает update
3. есть pushpage, который просто добавляет в конец очереди новую страницу
4. есть parsepage, который вызывается только из update. Он выдирает ссылки и для каждой ссылки вызывает pushpage

где ошибка в логике/архитектуре?

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

K>

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

HAS>>Уважаемые, встал в тупик и не знаю что делать, подскажите плз:

HAS>>Вопрос: с точки зрения многопоточности критичная для нас функция OnFinishThread, т.к. ее могут вызвать сразу несколько потоков одновременно, прально? Соотв. делаем, например, так:

HAS>>void OnFinishThread(Tthread *Thread)
HAS>>{
HAS>> EnterCriticalSection(&CS);
HAS>> try
HAS>> {
HAS>> if (Locker > 0) throw Exception("Херня, залезло слишком много потоков одновременно ");
HAS>> ++Locker;
HAS>> try
HAS>> {
HAS>> ParsePage(Thread->Page);
HAS>> }__finally
HAS>> {
HAS>> --Locker;
HAS>> };
HAS>> }__finally
HAS>> {
HAS>> LeaveCriticalSection(&CS);
HAS>> };
HAS>>};

HAS>>Подобный вариант НЕ РАБОТАЕТ ((

HAS>>Как подобное лучше сделать?!

K>Ведь при входе в OnFinishThread можно выставить семафор, а перед вызовом OnFinishThread его проверять и оценивать необходимость запуска.

да, можно. В принципе можно использовать любую методику синхронизации... Мне важно другое: с точки зрения синхронизации здесь вроде все правильно. Соотв. мне ВАЖНО ПОНЯТЬ, где что и почему не работает, дабы не наступать на эти же грабли в будущем!

Hello HAS, you wrote:

> в принципе так все и сделано:
> 1. есть update, который бегает по очереди и в зависимости от статуса страницы либо разбирает её, либо запускает новый поток
> 2. есть OnFinishthread, который вызывается по завершению любого потока, скачавшего страницу. он меняет статус страницы на скачанный (в рамках EnterCS & LeaveSC) и вызывает update
> 3. есть pushpage, который просто добавляет в конец очереди новую страницу
> 4. есть parsepage, который вызывается только из update. Он выдирает ссылки и для каждой ссылки вызывает pushpage
>
> где ошибка в логике/архитектуре?

В логике ошибок не замечено (при условии что везде производится синхронизированный доступ к одереди). Остаются ошибки реализации.

--
Всего хорошего, Слава
ICQ: 197577902

Hello HAS, you wrote:

> заманался гонять, такое ощущение, что отладчик глюкаво работает

Маловероятно.

> потому как очень часто выкидывает на код, где в принципе ничего быть не должно (там даже многопоточности нету)

Видимо в каком-то месте произошла порча памяти и что-нить подобное. Какая ошибка при этом возникает? AV?

--
Всего хорошего, Слава
ICQ: 197577902

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

HAS>Уважаемые, встал в тупик и не знаю что делать, подскажите плз:

HAS>Есть задача: многопоточная обработка данных. В частности: необходимо многопоточно анализировать веб страницы. С этих страниц мы выдираем ссылки и их анализируем и т.д.

HAS>Как мне это видится:
HAS>1. Есть поток, который занимается только скачиванием страницы в нужное место для последующего поиска. По завершению потока выполняется некая функция OnFinishThread,
HAS>2. Есть некая функция PushPage(address), которая добавляет страницу в общий список, присваивает странице address состояние типа NEED_DOWNLOAD и все, больше она ничего не делает.
HAS>3. Есть некая функция Update, которая просматривает список страниц и для каждой страницы со статусом NEED_DOWNLOAD запускает тот самый поток
HAS>4. Функция OnFinishThread берет ту страницу, которую скачал соотв. вызывающий поток, и вызывает функцию ParsePage, которая на каждый найденный урл вызовет соотв. PushAddress и т.д.
HAS>Соотв. функция Update циклическая и все скачивается, анализируется и т.д.

Вообще говоря, запускать на каждый адрес по потоку — это не лучшее решение. Легко можно попасть в ситуацию, когда будут одновременно запущены сотни и тысячи потоков, что приведет к перерасходу системных ресурсов (запуск потока — не очень дешевая операция, и синхронизация тысяч потоков для системы тоже дорога). Кроме того, будут еще и инициированы одновременно тысячи соединений (скачивание страниц), что тоже не есть хорошо.

Лучшим будет решение с пулом потоков, каждый из которых в цикле выбирает адреса страниц из глобального списка, скачивает страницу и анализирует результат. Соответственно, синхронизация нужна только при доступе к этому списку, ну и при помещении результата куда надо, конечно. Количество потоков легко регулировать в соответствии с загрузкой системой и количеством процессоров. Вариант — два пула потоков, один скачивает страницы, другой обрабатывает. Примерно так работают веб-сервера.

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

SA>Hello HAS, you wrote:

>> заманался гонять, такое ощущение, что отладчик глюкаво работает

SA>Маловероятно.

>> потому как очень часто выкидывает на код, где в принципе ничего быть не должно (там даже многопоточности нету)

SA>Видимо в каком-то месте произошла порча памяти и что-нить подобное. Какая ошибка при этом возникает? AV?

либо Исключение, либо просто выбрасывает в окно CPU Debug...
п.с. а что такое AV ?

SA>--
SA>Всего хорошего, Слава
SA>ICQ: 197577902