VPF::Помогите решить проблему с threads

FishHunter

Дата 12.2.2009, 15:35 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 47
Регистрация: 12.2.2009

Репутация: нет
Всего: нет

Добрый день Уважаемые,
Это мой первый пост поэтому немного о себе.
Решил я научиться кодить на перл, занимаюсь этим делом уже около 2 месяцев. Потихоньку дошел до потоков и застрял на такой проблеме. Вообщем есть скрипт, который в многопоточном режиме чекает ответы серверов по заданным урлам. Прошу не ругать за может быть корявость кода, но тем не менее ...

Код


#!/usr/bin/perl

use threads;
use LWP::UserAgent;


open(TEST,"<test.txt") or die ("Can not open file for reading : $!");
 my @urls=<TEST>;
close(TEST) or die ("Can not close file : $!");



my $start=time();

my $th_amount = 50; # количество потоков

if ( (scalar(@urls)%$th_amount) == 0 ){ my $iter=int(scalar(@urls)/$th_amount);}
else { $iter=(int(scalar(@urls)/$th_amount))+1;} # вычисляем сколько раз нужно запустить $th_amount потоков чтобы пройти весь массив @url

my $metka = 0;

for my $j (1..$iter) {
 my @threads=();
 for my $i (1..$th_amount) {  push @threads, threads->create(\&check_url, $i); }
 foreach my $thread (@threads) {  $thread->join(); }
 $metka=$metka+$th_amount; # определяем метку для начального индекса массива @url для следующей пачки потоков
}

my $end=time();
print $end-$start,"\tseconds past\n";





 sub check_url { 
    
    my $num = shift;
    my $index=$num-1+$metka; # определяем индекс массива @url для данного потока
    if ( $index > scalar(@urls) ) {return undef;}
    print "index ",$index," thread ", $num, " => ", time();
    my $browser = LWP::UserAgent->new;
    $browser->agent('Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 1.1.4322; .NET CLR 3.0.04506.30; .NET CLR 3.0.04506.648)');
    $browser->timeout(10);
    $browser->max_size(1024*1024/3);
    my $req = HTTP::Request->new(GET=>$urls[$index]);
    $req->header('Accept' => 'image/gif, image/x-xbitmap, image/jpeg, image/pjpeg, application/vnd.ms-powerpoint, application/vnd.ms-excel, application/msword, */*');
    $req->header("Accept-Language" => "en-US");
    my @ref=parse_URL ($urls[$index]);
    $req->header("Referer" => "$ref[0]://$ref[1]");
    my $resp = new HTTP::Response;
    $resp = $browser->request($req);
    my $status_code = $resp->code;
         
    if ($status_code == 200){
     my $html = $resp->content;
     if ($html=~/jazz/gi) {print "\tGood url and they talking about JAZZ\n";} else {print "\tGood url but they DO NOT know about jazz\n";}
     }
    else {print "\tError url\n";}
  }



sub parse_URL {
 my ($URL) = @_;
 (my @parsed =$URL =~ m@(\w+)://([^/:]+)(:\d*)?([^#]*)@) || return undef;
  if (defined $parsed[2]) {
    $parsed[2]=~ s/^://;
   }
  $parsed[3]='/' if ($parsed[0]=~/http/i && (length $parsed[3])==0);
  return @parsed if (defined $parsed[2]);
  $parsed[2] = 80;
  return @parsed;
}

на мой нубский взгляд, работа с потоками описана без ошибок, но тем не менее скрипт не хочет пахать,
вернее начинает, а потом вылетает с таким сообщением

index 1329 thread 30 => 1234445195      Error url
index 1301 thread 2 => 1234445194       Good url but they DO NOT know about jazz
index 1320 thread 21 => 1234445194      Error url
*** glibc detected *** perl: double free or corruption (fasttop): 0xb38047a0 ***
index 1300 thread 1 => 1234445194       Error url
index 1317 thread 18 => 1234445194      Error url
======= Backtrace: =========
/lib/libc.so.6[0x495fdd06]
/lib/libc.so.6(cfree+0x90)[0x496011e0]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_newCONSTSUB+0x154)[0x4985d254]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/auto/IO/IO.so(boot_IO+0x3d2)[0xb74eb4d2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_pp_entersub+0x40d)[0x4989641d]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_runops_standard+0x1f)[0x4988f88f]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x4982fffe]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_sv+0x5e6)[0x49834806]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_list+0x20b)[0x49834b2b]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_newATTRSUB+0x1152)[0x49868382]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_utilize+0x331)[0x498665e1]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_yyparse+0x1b22)[0x498577c2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x498c3200]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_pp_require+0xd52)[0x498c50b2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_runops_standard+0x1f)[0x4988f88f]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x4982fffe]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_sv+0x5e6)[0x49834806]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_list+0x20b)[0x49834b2b]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_newATTRSUB+0x1152)[0x49868382]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_utilize+0x331)[0x498665e1]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_yyparse+0x1b22)[0x498577c2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x498c3200]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_pp_require+0xd52)[0x498c50b2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_runops_standard+0x1f)[0x4988f88f]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x4982fffe]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_sv+0x5e6)[0x49834806]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_list+0x20b)[0x49834b2b]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_newATTRSUB+0x1152)[0x49868382]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_utilize+0x331)[0x498665e1]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_yyparse+0x1b22)[0x498577c2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x498c3200]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_pp_require+0xd52)[0x498c50b2]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_runops_standard+0x1f)[0x4988f88f]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so[0x4982fffe]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so(Perl_call_sv+0x5e6)[0x49834806]
/usr/lib/perl5/5.8.8/i386-linux-thread-multi/auto/threads/threads.so(Perl_ithread_run+0x1ae)[0xb7d28a5e]
/lib/libpthread.so.0[0x4970e45b]
/lib/libc.so.6(clone+0x5e)[0x49665e5e]
======= Memory map: ========
08048000-0804b000 r-xp 00000000 08:01 3260944    /usr/bin/perl
0804b000-0804c000 rw-p 00002000 08:01 3260944    /usr/bin/perl
0804c000-0b278000 rw-p 0804c000 00:00 0          [heap]
49577000-49591000 r-xp 00000000 08:01 53739634   /lib/ld-2.5.so
49591000-49592000 r--p 00019000 08:01 53739634   /lib/ld-2.5.so
49592000-49593000 rw-p 0001a000 08:01 53739634   /lib/ld-2.5.so
49595000-496d2000 r-xp 00000000 08:01 53739638   /lib/libc-2.5.so
496d2000-496d4000 r--p 0013d000 08:01 53739638   /lib/libc-2.5.so
496d4000-496d5000 rw-p 0013f000 08:01 53739638   /lib/libc-2.5.so
496d5000-496d8000 rw-p 496d5000 00:00 0
496da000-496dc000 r-xp 00000000 08:01 53739659   /lib/libdl-2.5.so
496dc000-496dd000 r--p 00001000 08:01 53739659   /lib/libdl-2.5.so
496dd000-496de000 rw-p 00002000 08:01 53739659   /lib/libdl-2.5.so
496e0000-49705000 r-xp 00000000 08:01 53739644   /lib/libm-2.5.so
49705000-49706000 r--p 00024000 08:01 53739644   /lib/libm-2.5.so
49706000-49707000 rw-p 00025000 08:01 53739644   /lib/libm-2.5.so
49709000-4971c000 r-xp 00000000 08:01 53739682   /lib/libpthread-2.5.so
4971c000-4971d000 r--p 00012000 08:01 53739682   /lib/libpthread-2.5.so
4971d000-4971e000 rw-p 00013000 08:01 53739682   /lib/libpthread-2.5.so
4971e000-49720000 rw-p 4971e000 00:00 0
49722000-49738000 r-xp 00000000 08:01 53741227   /lib/libselinux.so.1
49738000-4973a000 rw-p 00015000 08:01 53741227   /lib/libselinux.so.1
4973c000-49777000 r-xp 00000000 08:01 53741220   /lib/libsepol.so.1
49777000-49778000 rw-p 0003a000 08:01 53741220   /lib/libsepol.so.1
49778000-49782000 rw-p 49778000 00:00 0
49794000-49796000 r-xp 00000000 08:01 53739737   /lib/libutil-2.5.so
49796000-49797000 r--p 00001000 08:01 53739737   /lib/libutil-2.5.so
49797000-49798000 rw-p 00002000 08:01 53739737   /lib/libutil-2.5.so
497a4000-497ad000 r-xp 00000000 08:01 53739650   /lib/libcrypt-2.5.so
497ad000-497ae000 r--p 00008000 08:01 53739650   /lib/libcrypt-2.5.so
497ae000-497af000 rw-p 00009000 08:01 53739650   /lib/libcrypt-2.5.so
497af000-497d6000 rw-p 497af000 00:00 0
497d8000-497eb000 r-xp 00000000 08:01 53739729   /lib/libnsl-2.5.so
497eb000-497ec000 r--p 00012000 08:01 53739729   /lib/libnsl-2.5.so
497ec000-497ed000 rw-p 00013000 08:01 53739729   /lib/libnsl-2.5.so
497ed000-497ef000 rw-p 497ed000 00:00 0
497f1000-49800000 r-xp 00000000 08:01 53739731   /lib/libresolv-2.5.so
49800000-49801000 r--p 0000e000 08:01 53739731   /lib/libresolv-2.5.so
49801000-49802000 rw-p 0000f000 08:01 53739731   /lib/libresolv-2.5.so
49802000-49804000 rw-p 49802000 00:00 0
4980e000-49939000 r-xp 00000000 08:01 3279280    /usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so
49939000-4993e000 rw-p 0012a000 08:01 3279280    /usr/lib/perl5/5.8.8/i386-linux-thread-multi/CORE/libperl.so
4993e000-49940000 rw-p 4993e000 00:00 0
4999b000-499c8000 r-xp 00000000 08:01 3279415    /usr/lib/libgssapi_krb5.so.2.2
499c8000-499c9000 rw-p 0002d000 08:01 3279415    /usr/lib/libgssapi_krb5.so.2.2
499cb000-499d3000 r-xp 00000000 08:01 3279288    /usr/lib/libkrb5support.so.0.1
499d3000-499d4000 rw-p 00007000 08:01 3279288    /usr/lib/libkrb5support.so.0.1
499d6000-499fb000 r-xp 00000000 08:01 3279375    /usr/lib/libk5crypto.so.3.1
499fb000-499fc000 rw-p 00025000 08:01 3279375    /usr/lib/libk5crypto.so.3.1
499fe000-49a00000 r-xp 00000000 08:01 53741240   /lib/libkeyutils-1.2.so
49a00000-49a01000 rw-p 00001000 08:01 53741240   /lib/libkeyutils-1.2.so
49a03000-49b20000 r-xp 00000000 08:01 53741257   /lib/libcrypto.so.0.9.8b
49b20000-49b33000 rw-p 0011c000 08:01 53741257   /lib/libcrypto.so.0.9.8b
49b33000-49b36000 rw-p 49b33000 00:00 0
49b38000-49b79000 r-xp 00000000 08:01 53741268   /lib/libssl.so.0.9.8b
49b79000-49b7d000 rw-p 00040000 08:01 53741268   /lib/libssl.so.0.9.8b
83800000-838f8000 rw-p 83800000 00:00 0
838f8000-83900000 ---p 838f8000 00:00 0
83a00000-83a22000 rw-p 83a00000 00:00 0
83a22000-83b00000 ---p 83a22000 00:00 0
84400000-844db000 rw-p 84400000 00:00 0
844db000-84500000 ---p 844db000 00:00 0
.......
93a00000-93b00000 rw-p 93a00000 00:00 0
93b00000-93c00000 rw-p 93b00000 Aborted
[root@server myscripts]#

вопрос: что сие значит и как это победить? или можно забить на потоки (читал неоднократно что в них полно глюков) и решать подобные задачи форками? (их ещё не учил smile

)

PS забыл добавить, что с малым кол-вом данных все работает на ура, но если я в @urls загоню скажем 6к элементов то уже косяк :(

Это сообщение отредактировал(а) FishHunter - 12.2.2009, 16:08

ginnie

Дата 12.2.2009, 15:43 (ссылка)

(нет голосов)

Загрузка ...

Эксперт

Профиль
Группа: Комодератор
Сообщений: 1287
Регистрация: 6.1.2008
Где: Москва

Репутация: 38
Всего: 49

FishHunter, Вашу задачу можно решить при помощи событийной машины, посмотрите в сторону модуля EV. Если покажется непонятным, делайте через процессы.

--------------------

Написать код, понятный компьютеру, может каждый, но только хорошие программисты пишут код, понятный людям. (Мартин Фаулер. Рефакторинг)

FishHunter

Дата 12.2.2009, 16:04 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 47
Регистрация: 12.2.2009

Репутация: нет
Всего: нет

Я догадываюсь, что данную задачу можно решить по-другому, обязательно посмотрю модуль EV. Но смысл не в том чтобы решить именно эту задачу, а смысл в том чтобы решить её при помощи потоков или по-крайней мере определиться что потоками её на данном этапе развития модуля threads не решить

tolkien

Дата 12.2.2009, 19:07 (ссылка)

(нет голосов)

Загрузка ...

Опытный

Профиль
Группа: Участник
Сообщений: 277
Регистрация: 5.4.2008

Репутация: 4
Всего: 4

В самом деле сделайте через процессы. Переделывать практически ничего не придется. Зато если будет работать на ура. Значит кривые потоки.

FishHunter

Дата 12.2.2009, 20:15 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 47
Регистрация: 12.2.2009

Репутация: нет
Всего: нет

Цитата(tolkien @ 12.2.2009, 19:07)

ок попробую, токо для меня "практически ничего" значит все smile

PS что-то модуль EV совсем пытается взорвать мне мозг smile

PSS победил я сию ошибку, достаточно было буфер вывода почистить STDOUT->autoflush(1);
но теперь скрипт вылетает (после долгой работы, раз в 5 дольше работает чем раньше) с сообщением Alarm clock :(
это типа сигнал ALRM от системы? ставил $SIG{ALRM} = 'IGNORE'; не помогло :(

Это сообщение отредактировал(а) FishHunter - 12.2.2009, 23:46

NuINu

Дата 13.2.2009, 12:08 (ссылка)

(голосов:1)

Загрузка ...

Шустрый

Профиль
Группа: Участник
Сообщений: 131
Регистрация: 19.7.2008

Репутация: 5
Всего: 6

FishHunter, на мой взгляд вы не правильно используете потоки, что это за фигня создаем 50 потоков, потом ждем пока они завершатья, потом еще создаем 50 потоков и т.д.
потоки должны РАБОТАТЬ!
нужно создать 50 потоков и ВСЕ! дальше пусть они работают, а ваше дело расскидывать задания для этих потоков.

вот пример как правильно

Код


#!/usr/bin/perl

use strict;
#Не работает если перл собран без поддержики
use threads;
use threads::shared;
use Data::Dumper;
use IO::Handle;

    #прочитаем массив ссылок из файла, предназначенных для обработки
open(TEST,"<test.txt") or die ("Can not open file for reading : $!");
my @urls=<TEST>;
close(TEST) or die ("Can not close file : $!");

    #чтобы вводилось сразу
STDOUT->autoflush(1);
    #Создадим разделяемый индекс указывающий на следующий элемент массива для обработки
my $sh_ind = 0;
share($sh_ind);
    #Создадим массив потоков которые будут обрабатывать наш массив
my $th_amount = 10;
foreach (0..$th_amount) {
    threads->create(\&check_url, \@urls);
}
print "Main: make all threads!\n";

    ### Collect the bits and pieces! ...
$_->join foreach threads->list;
print "Main: END!\n";

    #процедура работы дочерних потоков
sub check_url { 
    my $p_arr = shift;
    
    my $tid    = threads->tid();
    print "Thread ($tid): START\n";
    my $max_ind = $#$p_arr;
    my $time_wait;
    my $work_ind;
    while(1) {
    {
        lock($sh_ind);
        $work_ind = $sh_ind;
        if($work_ind <= $max_ind) {
        $sh_ind++;    #увеличиваем индекс для последующей обработки 
        }
    }
    print "Thread ($tid): index for work = $work_ind\n";
    if($work_ind > $max_ind) {    #Завершим работу если индекс стал больше чем количество элементов
        last
    };
        #вместо реальной работы просто засыпаем на произвольное число секунд от 1 до 3
    $time_wait = 1+int(rand 3);
    print "Thread ($tid): w_time=$time_wait, urls: $p_arr->[$work_ind]";
    sleep  $time_wait;
    }
    
    print "Thread ($tid): END\n";
}

ginnie

Дата 13.2.2009, 20:53 (ссылка)

(голосов:1)

Загрузка ...

Эксперт

Профиль
Группа: Комодератор
Сообщений: 1287
Регистрация: 6.1.2008
Где: Москва

Репутация: 38
Всего: 49

FishHunter, наткнулся на обсуждение использования потоков в Perl, почитайте.

--------------------

FishHunter

Дата 13.2.2009, 23:12 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 47
Регистрация: 12.2.2009

Репутация: нет
Всего: нет

Цитата(ginnie @ 13.2.2009, 20:53)

FishHunter, наткнулся на обсуждение использования потоков в Perl, почитайте.

судя по этому разговору на нитки в перл можно забить smile

Добавлено через 4 минуты и 53 секунды

Цитата(NuINu @ 13.2.2009, 12:08)

Огромый респект! и большое человеческое спасибо smile

я так и знал что саму раздачу данных потокам организовал через одно место.
Но проблемка alarm clock осталась, причем походу это косяк LWP, а точнее взаимодействия LWP и threads, т.к. если не пользовать LWP все пашет на ура. Есть идеи что это может быть? гугл ничего внятного что-то мне не сказал :(

perloid

Дата 14.2.2009, 03:17 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 33
Регистрация: 14.11.2008

Репутация: нет
Всего: -1

так заюзайте curl c параллельными запросами, есть модуль для него.
http://search.cpan.org/~crisb/WWW-Curl-3.0...W/Curl/Multi.pm
намного быстрее будет работать, чем городить огород с нитями.

+ существует альтернативный модуль LWP::Parallel

Это сообщение отредактировал(а) perloid - 14.2.2009, 03:29

NuINu

Дата 14.2.2009, 15:22 (ссылка)

(нет голосов)

Загрузка ...

Шустрый

Профиль
Группа: Участник
Сообщений: 131
Регистрация: 19.7.2008

Репутация: 5
Всего: 6

Цитата(FishHunter @ 13.2.2009, 21:12

)

Но проблемка alarm clock осталась, причем походу это косяк LWP, а точнее взаимодействия LWP и threads,

я сам не пробовал, но вам посоветую попробовать(все равно больше ничего не остается как переделывать).
модуль use Thread::Signal;
скорее всего он будет работать в версиях 5.8.9 и 5.10

у меня 5.8.8 в нем в тредах нет такйо фишки как thr->kill, а значит нельзя будет передать сигнал в тред правильно.

но вы попробуйте..

ЗЫ: не понимаю столь трепетной любви в людях к тредам... многие ссылаются на совместимость с виндой, в гробу я эту совместимость видел, в винде по определению ничего не должно работать!!! юзайте fork

NuINu

Дата 14.2.2009, 23:11 (ссылка)

(голосов:2)

Загрузка ...

Шустрый

Профиль
Группа: Участник
Сообщений: 131
Регистрация: 19.7.2008

Репутация: 5
Всего: 6

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

Код


#!/usr/bin/perl -w

use IO::File;
use POSIX 'WNOHANG';
use strict;
use Data::Dumper;

#Мультипроцессная программа-пример, простейшая система состоящая из главного процесса, и нескольких подчиненных 
#процессов - рабочих, главный процесс порождает рабочих, дает им один общий канал для отчетов(для передачи данных
#от рабочего - главному) и индивидуальные каналы для передачи данных от главного к каждому конкретному рабочему
# для получения задания рабочим. Создав рабочих главный начинает раздавать команды.
# После создания рабочий обращается к главному с просьбой дать задание, по завершении которого передает ответ главному
# Главный получает отчеты и раздает приказы.

my $max_workers = 10;
my $quit = 0;
my %childs;
my $cnt_worker;

my $w_ready = 1;    #готов к работе
my $w_end_w = 2;    #закончил работу, обработал

my $b_wrk   = 1;    #обработай
my $b_cls   = 2;    #завершись

my $deb1    = 0;    #вывод отладочной информации 1

    #Прочитаем список URLs которые необходимо обработать
    #прочитаем массив ссылок из файла, предназначенных для обработки
open(TEST,"<test.txt") or die ("Can not open file for reading : $!");
my @urls=<TEST>;
map {chomp} @urls;
close(TEST) or die ("Can not close file : $!");

    #Установим обработчиков сигналов
$SIG{PIPE} = 'IGNORE';
$SIG{CHLD} = 'sig_child';
    #может быть здесь стоит убить и порожденные процессы  или они сами умрут когда умрет лидер?
$SIG{TERM} = $SIG{INT} = sub{$quit++};

    #Порождаем процессы которые будут выполнять основную работу и докладывать "хозяину", 
    #хозяин же, будет давать работу! через индивидуальные командные каналы
my  ($res_mk_cmd, $fh_boss, $fh_worker);
pipe(READER, WRITER) or die "pipe no good: $!";
$| = 1;
my($i);
my $child;
for($i = 0; $i < $max_workers; $i++) {
    #При создании worker создаем индивидуальный командный канал
    $fh_boss = undef;                #ЭТО очень ОБЯЗАТЕЛЬНОЕ действие!
    $res_mk_cmd = pipe($fh_worker, $fh_boss);
    die "Can't make command pipe: $!" unless defined $res_mk_cmd;
    $child = fork();
    die "Can't fork: $!" unless defined $child;
    if ($child == 0) { #Work inchild process
    close $fh_boss;
    close READER;
    do_child($fh_worker);
    exit(0);
    }
    #Установим не буферизированный вывод при передаче команд worker процессу
    select $fh_boss; $| = 1;
    select STDOUT;
    close $fh_worker;
    #Сохраним данные о child процессе в хеше 
    $childs{$child} = {id=>$i, fh_cmd=>$fh_boss};  
    print "Main: Build child pid $child id $i\n" if $deb1;
}
close WRITER;

print "Main: I build $i worker\n" if $deb1;
$cnt_worker = $i;

    #Все процессы рабочие готовы, начинаем раздачу заданий и прием ответов
my $cur_id_urls = 0;
my $in;
my ($id_rep,$mess);
my ($m_id, $m_res);
    #Будем работать 
    #пока ненадо выйти, есть рабочие, нет ошибок чтения отчетов от рабочих, и пока не обработали все ссылки
while (!$quit and ($cnt_worker > 0) and defined ($in = <READER>)) {
        # and ($cur_id_urls =< $#urls)
        #Итак принимаем доклад
    print "Main: get report: $in"  if $deb1;
    chomp $in;
    ($child, $id_rep, $mess) = split(/:/, $in);
    #Проверим есть ли у нас вообще такой работник
    if(!defined($childs{$child})) {    #Нет? ужас!!! откуда он взялся!
    print "Main: get report from unknown child($child)!\n" if $deb1;
    next;
    }    
    #Расшифровываем о чем этот доклад
    if($id_rep == $w_end_w) {
    print "Main: childs($child), reported end work!\n" if $deb1;
    ($m_id, $m_res) = split(/,/, $mess);
    #print "Main: URLs($m_id)='$urls[$m_id]', is '$m_res'\n";
    print "($child): URLs($m_id)='$urls[$m_id]', is '$m_res'\n";
    } elsif($id_rep == $w_ready) {
    print "Main: child($child), reported ready work!\n" if $deb1;
    } else {
    print "Main: child($child), send bad report '$id_rep', with message '$mess'\n" if $deb1;
    }
    #Даем задание рабочему если в этом есть необходимость
    my $fh = $childs{$child}->{fh_cmd};
    if($cur_id_urls <= $#urls) { #Есть еще не обработанные задания?
    print "Main($$): put new cmd 'JOB' to worker($child)\n" if $deb1;
    print $fh "$b_wrk:$cur_id_urls\n";
    $cur_id_urls++;
    } else {    #Нет? пошлем рабочему комаду завершиться
    print "Main($$): put new cmd 'END' to worker($child)\n" if $deb1;
    print $fh "$b_cls\n";
    }
}

    #Определим причину по которой завершилась программа
if($quit) {
    print "Exit by quit = $quit\n";
    print "cnt_worker is $cnt_worker\n";
}

if($cnt_worker <= 0) {
    print "Exit by all childs ended!\n";
    print "quit is $quit\n";
    print "cnt_worker is $cnt_worker\n";
}

if($cur_id_urls > $#urls) {
    print "Exit by cur_id_urls($cur_id_urls) gt count urls !\n";
}

#Если остались незавершенные процессы рабочие убъем их, благо информация что это за процессы у нас есть
while($cnt_worker > 0) {
    print "Main: Run worker killer!\n" if $deb1;
    foreach $child (keys %childs) {
    if($childs{$child}->{id} != 0) {
        print "Main: Please child($child) - kill self!\n" if $deb1;
        kill ('TERM', $child);
    }
    }
    sleep 1;
}

print "Main: --------------- A parent End print---------------------\n";
exit(0);


    #Обработка сигнала завершения дочернего процесса
sub sig_child {
    my $child; 
    while(($child = waitpid(-1, WNOHANG)) > 0) {
    my $id = $childs{$child}->{id}; 
    if(defined($id)) {
        #можно пометить процесс как убитый, а можно и ничего не делать
        $childs{$child}->{id} = 0;
        print "Kiled child id $id, PID($child)\n" if $deb1;
    } else {
        print "Kiled unknown child PID($child)\n" if $deb1;
    }
    $cnt_worker--;
    }
}


    #Процедура в которой выполняется работа дочерним процессом
sub do_child {
    my $fh = shift;
    my $quit_child = 0;
    my $timeout    = 5;
    my $job;
    my $res;
    my ($id_cmd, $work);
    #Установим собственный обработчик сигнала завершения процесса
    $SIG{TERM} = $SIG{INT} = sub{$quit_child++};
        #Сделаем так что бы отчет передавался главному процессу без буферизации
    select WRITER; $| = 1;
    select STDOUT;
    #Сообщим что рабочий готов к работе
    print WRITER "$$:$w_ready\n";
    while($quit_child == 0) {
        #Читаем команду от главного
    $job = <$fh>;
        #Задание получено, отработаем
    chomp $job;
    ($id_cmd, $work) = split(/:/, $job);
    if($id_cmd == $b_wrk) {        #Получена команда обработать элемент массива urls, $work-это индекс в urls
        #print "Child($$): work urls($work): '$urls[$work]'!\n";
        #Опять же вместо реальной работы некоторое время ждем от 1 до 2 секунд
        sleep(1+int(rand(2)));    
        #и возвращаем случайный результат где-то ошибка 1 из 5.
        $res  = (int(rand(6))) ? 'ok' : 'bad';
        print WRITER "$$:$w_end_w:$work,$res\n";
    } elsif ($id_cmd == $b_cls) {    #Получена команда завершения работы
        #print "Child($$): get command terminated!!!\n";
        $quit_child++;
    } else {
        print "Child($$): get unknown terminated!!!\n";
    }
    }
    
    close $fh;
    close WRITER;
    print "Child($$): Ended!\n" if $deb1;
}

FishHunter

Дата 18.2.2009, 11:15 (ссылка)

(нет голосов)

Загрузка ...

Новичок

Профиль
Группа: Участник
Сообщений: 47
Регистрация: 12.2.2009

Репутация: нет
Всего: нет

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

mvsgt

Дата 31.5.2009, 11:24 (ссылка)

(нет голосов)

Загрузка ...

Бывалый

Профиль
Группа: Участник
Сообщений: 209
Регистрация: 27.3.2009

Репутация: 1
Всего: 1

Насчёт потоков. В современных версиях Perl используется threads вместо Thread и это принципиально.

Код


For new code the use of the Thread module is discouraged and the direct use 
of the threads and threads::shared modules is encouraged instead.

Насчёт LWP и потоков: с threads LWP работает. А вот с LWP::Parallel я нарвался на проблемы при использовании прокси - прокси-запросы выполнены блокирующими запросами, и вся параллельность пропадает.

Единственная беда с потоками - памяти уходит много, поэтому 100 потоков - это уже много.

kukich

Дата 10.2.2011, 16:20 (ссылка)

(нет голосов)

Загрузка ...

Шустрый

Профиль
Группа: Участник
Сообщений: 69
Регистрация: 2.12.2009

Репутация: нет
Всего: нет

Цитата(NuINu @ 14.2.2009, 23:11)

Код


#!/usr/bin/perl -w

use IO::File;
use POSIX 'WNOHANG';
use strict;
use Data::Dumper;

#Мультипроцессная программа-пример, простейшая система состоящая из главного процесса, и нескольких подчиненных 
#процессов - рабочих, главный процесс порождает рабочих, дает им один общий канал для отчетов(для передачи данных
#от рабочего - главному) и индивидуальные каналы для передачи данных от главного к каждому конкретному рабочему
# для получения задания рабочим. Создав рабочих главный начинает раздавать команды.
# После создания рабочий обращается к главному с просьбой дать задание, по завершении которого передает ответ главному
# Главный получает отчеты и раздает приказы.

my $max_workers = 10;
my $quit = 0;
my %childs;
my $cnt_worker;

my $w_ready = 1;    #готов к работе
my $w_end_w = 2;    #закончил работу, обработал

my $b_wrk   = 1;    #обработай
my $b_cls   = 2;    #завершись

my $deb1    = 0;    #вывод отладочной информации 1

    #Прочитаем список URLs которые необходимо обработать
    #прочитаем массив ссылок из файла, предназначенных для обработки
open(TEST,"<test.txt") or die ("Can not open file for reading : $!");
my @urls=<TEST>;
map {chomp} @urls;
close(TEST) or die ("Can not close file : $!");

    #Установим обработчиков сигналов
$SIG{PIPE} = 'IGNORE';
$SIG{CHLD} = 'sig_child';
    #может быть здесь стоит убить и порожденные процессы  или они сами умрут когда умрет лидер?
$SIG{TERM} = $SIG{INT} = sub{$quit++};

    #Порождаем процессы которые будут выполнять основную работу и докладывать "хозяину", 
    #хозяин же, будет давать работу! через индивидуальные командные каналы
my  ($res_mk_cmd, $fh_boss, $fh_worker);
pipe(READER, WRITER) or die "pipe no good: $!";
$| = 1;
my($i);
my $child;
for($i = 0; $i < $max_workers; $i++) {
    #При создании worker создаем индивидуальный командный канал
    $fh_boss = undef;                #ЭТО очень ОБЯЗАТЕЛЬНОЕ действие!
    $res_mk_cmd = pipe($fh_worker, $fh_boss);
    die "Can't make command pipe: $!" unless defined $res_mk_cmd;
    $child = fork();
    die "Can't fork: $!" unless defined $child;
    if ($child == 0) { #Work inchild process
    close $fh_boss;
    close READER;
    do_child($fh_worker);
    exit(0);
    }
    #Установим не буферизированный вывод при передаче команд worker процессу
    select $fh_boss; $| = 1;
    select STDOUT;
    close $fh_worker;
    #Сохраним данные о child процессе в хеше 
    $childs{$child} = {id=>$i, fh_cmd=>$fh_boss};  
    print "Main: Build child pid $child id $i\n" if $deb1;
}
close WRITER;

print "Main: I build $i worker\n" if $deb1;
$cnt_worker = $i;

    #Все процессы рабочие готовы, начинаем раздачу заданий и прием ответов
my $cur_id_urls = 0;
my $in;
my ($id_rep,$mess);
my ($m_id, $m_res);
    #Будем работать 
    #пока ненадо выйти, есть рабочие, нет ошибок чтения отчетов от рабочих, и пока не обработали все ссылки
while (!$quit and ($cnt_worker > 0) and defined ($in = <READER>)) {
        # and ($cur_id_urls =< $#urls)
        #Итак принимаем доклад
    print "Main: get report: $in"  if $deb1;
    chomp $in;
    ($child, $id_rep, $mess) = split(/:/, $in);
    #Проверим есть ли у нас вообще такой работник
    if(!defined($childs{$child})) {    #Нет? ужас!!! откуда он взялся!
    print "Main: get report from unknown child($child)!\n" if $deb1;
    next;
    }    
    #Расшифровываем о чем этот доклад
    if($id_rep == $w_end_w) {
    print "Main: childs($child), reported end work!\n" if $deb1;
    ($m_id, $m_res) = split(/,/, $mess);
    #print "Main: URLs($m_id)='$urls[$m_id]', is '$m_res'\n";
    print "($child): URLs($m_id)='$urls[$m_id]', is '$m_res'\n";
    } elsif($id_rep == $w_ready) {
    print "Main: child($child), reported ready work!\n" if $deb1;
    } else {
    print "Main: child($child), send bad report '$id_rep', with message '$mess'\n" if $deb1;
    }
    #Даем задание рабочему если в этом есть необходимость
    my $fh = $childs{$child}->{fh_cmd};
    if($cur_id_urls <= $#urls) { #Есть еще не обработанные задания?
    print "Main($$): put new cmd 'JOB' to worker($child)\n" if $deb1;
    print $fh "$b_wrk:$cur_id_urls\n";
    $cur_id_urls++;
    } else {    #Нет? пошлем рабочему комаду завершиться
    print "Main($$): put new cmd 'END' to worker($child)\n" if $deb1;
    print $fh "$b_cls\n";
    }
}

    #Определим причину по которой завершилась программа
if($quit) {
    print "Exit by quit = $quit\n";
    print "cnt_worker is $cnt_worker\n";
}

if($cnt_worker <= 0) {
    print "Exit by all childs ended!\n";
    print "quit is $quit\n";
    print "cnt_worker is $cnt_worker\n";
}

if($cur_id_urls > $#urls) {
    print "Exit by cur_id_urls($cur_id_urls) gt count urls !\n";
}

#Если остались незавершенные процессы рабочие убъем их, благо информация что это за процессы у нас есть
while($cnt_worker > 0) {
    print "Main: Run worker killer!\n" if $deb1;
    foreach $child (keys %childs) {
    if($childs{$child}->{id} != 0) {
        print "Main: Please child($child) - kill self!\n" if $deb1;
        kill ('TERM', $child);
    }
    }
    sleep 1;
}

print "Main: --------------- A parent End print---------------------\n";
exit(0);


    #Обработка сигнала завершения дочернего процесса
sub sig_child {
    my $child; 
    while(($child = waitpid(-1, WNOHANG)) > 0) {
    my $id = $childs{$child}->{id}; 
    if(defined($id)) {
        #можно пометить процесс как убитый, а можно и ничего не делать
        $childs{$child}->{id} = 0;
        print "Kiled child id $id, PID($child)\n" if $deb1;
    } else {
        print "Kiled unknown child PID($child)\n" if $deb1;
    }
    $cnt_worker--;
    }
}


    #Процедура в которой выполняется работа дочерним процессом
sub do_child {
    my $fh = shift;
    my $quit_child = 0;
    my $timeout    = 5;
    my $job;
    my $res;
    my ($id_cmd, $work);
    #Установим собственный обработчик сигнала завершения процесса
    $SIG{TERM} = $SIG{INT} = sub{$quit_child++};
        #Сделаем так что бы отчет передавался главному процессу без буферизации
    select WRITER; $| = 1;
    select STDOUT;
    #Сообщим что рабочий готов к работе
    print WRITER "$$:$w_ready\n";
    while($quit_child == 0) {
        #Читаем команду от главного
    $job = <$fh>;
        #Задание получено, отработаем
    chomp $job;
    ($id_cmd, $work) = split(/:/, $job);
    if($id_cmd == $b_wrk) {        #Получена команда обработать элемент массива urls, $work-это индекс в urls
        #print "Child($$): work urls($work): '$urls[$work]'!\n";
        #Опять же вместо реальной работы некоторое время ждем от 1 до 2 секунд
        sleep(1+int(rand(2)));    
        #и возвращаем случайный результат где-то ошибка 1 из 5.
        $res  = (int(rand(6))) ? 'ok' : 'bad';
        print WRITER "$$:$w_end_w:$work,$res\n";
    } elsif ($id_cmd == $b_cls) {    #Получена команда завершения работы
        #print "Child($$): get command terminated!!!\n";
        $quit_child++;
    } else {
        print "Child($$): get unknown terminated!!!\n";
    }
    }
    
    close $fh;
    close WRITER;
    print "Child($$): Ended!\n" if $deb1;
}

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

NuINu

Дата 10.2.2011, 17:55 (ссылка)

(нет голосов)

Загрузка ...

Шустрый

Профиль
Группа: Участник
Сообщений: 131
Регистрация: 19.7.2008

Репутация: 5
Всего: 6

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

0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема \| Perl: Общие вопросы \| Следующая тема »