🚀 go-pugleaf

Krzysztof Lewicki wrote:
> Potrzebuje użyć wątków perlowych i natrafiłem na pewien problem. Napisałem 
> sobie taki oto przykładowy fragment kodu:
> 
> my $queue = Thread::Queue->new();
> my $t1 = threads->create(\&test_reader);
> my $t2 = threads->create(\&test_writer);
> 
> $t2->join();
> 
> sub test_writer
> {
> 	while (1)
> 	{
> 		sleep 1;
> 		my %t = (1 => "A", 2 => "B")
> 		$queue->enqueue(share(%t));
> 	}	
> }
> 
> sub test_reader
> {
> 	threads->detach();
> 	
> 	while (my $ref= $queue->dequeue())
> 	{
> 		my %hash = %{$ref}; # wyswietla np. HASH(0x1110970)
> 		print %hash; # tu juz nic nie wyswietla
> 	}
> }
> 
> Jedna kolejka, 2 watki, jeden czyta z kolejki, drugi zapisuje. Niby 
> proste. Problem pojawia sie jednak w funkcji test_reader. Otóż podczas 
> wyciagania z kolejki dostaje referencje na hash (print $ref daje cos w 
> stylu HASH(0x1110970)) i dotad - chyba - jest OK. Kiedy probuje wyciagnac 
> hasha z referencji dostaję... pustkę.
> 
> Nie wiem co z tym fantem zrobic, moze ktos mnie naprowadzi?

Mnie zazwyczaj sporo wyjaśnia:
use Dumper;
print Dumper $referencja;

-- 
P.M.

On Wed, 23 Dec 2009 16:40:36 +0000 (UTC), Krzysztof Lewicki wrote:

> 	while (1)
> 	{
> 		sleep 1;
> 		my %t = (1 => "A", 2 => "B")
> 		$queue->enqueue(share(%t));
> 	}	

Zadeklarowałeś %t jako lokalny w obrębie pętli. Należało zrobić tak:
        my %t;
	while (1)
	{
		sleep 1;
		%t = (1 => "A", 2 => "B");
		$queue->enqueue(share(%t));
	}	

(zakładam, że ten zgubiony średnik to błąd przy przeklejaniu...)

Zupełnie odrębną kwestią jest, czemu używasz join() na $t2, skoro explicite
uczyniłeś go "unjoinable" za pomocą detach().

-- 
Szymon Sokół (SS316-RIPE) -- Network Manager                                 B
Computer Center, AGH - University of Science and Technology, Cracow, Poland  O
http://home.agh.edu.pl/szymon/ PGP key id: RSA: 0x2ABE016B, DSS: 0xF9289982  F
Free speech includes the right not to listen, if not interested -- Heinlein  H

Paweł Matejski wrote:
> Krzysztof Lewicki wrote:
>>
>> Jedna kolejka, 2 watki, jeden czyta z kolejki, drugi zapisuje. Niby 
>> proste. Problem pojawia sie jednak w funkcji test_reader. Otóż podczas 
>> wyciagania z kolejki dostaje referencje na hash (print $ref daje cos w 
>> stylu HASH(0x1110970)) i dotad - chyba - jest OK. Kiedy probuje wyciagnac 
>> hasha z referencji dostaję... pustkę.
>>
>> Nie wiem co z tym fantem zrobic, moze ktos mnie naprowadzi?
> 
> Mnie zazwyczaj sporo wyjaśnia:
> use Dumper;
> print Dumper $referencja;

Ach, dopiero teraz mnie oświeciło - powyższa rada niewiele Ci da.

Shared zwraca referencje i tylko tą referencję (czyli sam adres) współdzieli miedzy wątkami, dane są gubione.
Ale według dokumentacji do Thread::Queue, nie potrzeba używać share - jeśli zmienna nie jest dzielona, kolejka powinna
sklonować dane.

perldoc Thread::Queue
If not already thread-shared, the other complex data types will be cloned (recursively, if needed, and including any
"bless"ings and read-only settings) into thread-shared structures before being placed onto a queue.


-- 
P.M.

Krzysztof Lewicki <menth0l@remove.this.o2.pl> pisze:
> Potrzebuje użyć wątków perlowych i natrafiłem na pewien problem. Napisałem 
> sobie taki oto przykładowy fragment kodu:
>
> my $queue = Thread::Queue->new();
> my $t1 = threads->create(\&test_reader);
> my $t2 = threads->create(\&test_writer);
>
> $t2->join();
>
> sub test_writer
> {
> 	while (1)
> 	{
> 		sleep 1;
> 		my %t = (1 => "A", 2 => "B")
> 		$queue->enqueue(share(%t));
> 	}	
> }
>
> sub test_reader
> {
> 	threads->detach();
> 	
> 	while (my $ref= $queue->dequeue())
> 	{
> 		my %hash = %{$ref}; # wyswietla np. HASH(0x1110970)
> 		print %hash; # tu juz nic nie wyswietla
> 	}
> }
>
> Jedna kolejka, 2 watki, jeden czyta z kolejki, drugi zapisuje. Niby 
> proste. Problem pojawia sie jednak w funkcji test_reader. Otóż podczas 
> wyciagania z kolejki dostaje referencje na hash (print $ref daje cos w 
> stylu HASH(0x1110970)) i dotad - chyba - jest OK. Kiedy probuje wyciagnac 
> hasha z referencji dostaję... pustkę.
>
> Nie wiem co z tym fantem zrobic, moze ktos mnie naprowadzi?

A) Spróbuj z odwróceniem kolejności operacji:
najpierw "share" hash, potem "upychaj" dane w hashu, a na  końcu
przesyłaj referencje do hasha
  my %t :shared = (1 => "A", 2 => "B")
  $queue->enqueue(\%t)
*ALBO*
B) spróbuj modułu Thread::Queue::Any z CPAN.org [nie używałem]

<quote src="man Threads::shared">
  BUGS AND LIMITATIONS
    When "share" is used on arrays, hashes, array refs or hash refs,
    any data they contain will be lost.
</quote>

P.S
Ja się "naciąłem" na to jak chciałem przesyłać referencje do tablic
przez kolejki między wątkami. To jest kolejna sprawa "proste po" :-)

-- 
Andrzej Adam Filip : anfi@onet.eu : Andrzej.Filip@gmail.com
Od jednego pocałunku kobieta się nie zepsuje.
  -- Przysłowie gruzińskie (pl.wikiquote.org)
[ http://groups.google.pl/groups?selm=iqrpawesz7-9CN@cynthia.brudna.chmurka.net ]

On Wed, 23 Dec 2009 19:49:33 +0100, Andrzej Adam Filip wrote:

> Krzysztof Lewicki <menth0l@remove.this.o2.pl> pisze:
[----]
> <quote src="man Threads::shared">
>   BUGS AND LIMITATIONS
>     When "share" is used on arrays, hashes, array refs or hash refs,
>     any data they contain will be lost.
> </quote>

Ja bym w ogóle ten kod przerobił tak:

use threads;
use threads::shared;
use Thread::Queue;

my $queue = Thread::Queue->new();
my $t1 = threads->create(\&test_reader);
my $t2 = threads->create(\&test_writer);

$t1->join();
$t2->join();

sub test_writer
{
  my %t;
  my $rt = share(%t);
  while(1)
  {
    sleep 1;
    %t = (1 => 'A', 2 => 'B');
    $queue->enqueue($rt);
  }
}

sub test_reader
{
  my %hash;
  my $ref;
  while($ref=$queue->dequeue())
  {
    %hash = %{$ref}; 
    print %hash, "\n"; 
  }
}

U mnie to działa, sprawdzone.
-- 
Szymon Sokół (SS316-RIPE) -- Network Manager                                 B
Computer Center, AGH - University of Science and Technology, Cracow, Poland  O
http://home.agh.edu.pl/szymon/ PGP key id: RSA: 0x2ABE016B, DSS: 0xF9289982  F
Free speech includes the right not to listen, if not interested -- Heinlein  H

Dnia 23-12-2009 o 17:40:36 Krzysztof Lewicki <menth0l@remove.this.o2.pl>  
napisał(a):

> Potrzebuje użyć wątków perlowych i natrafiłem na pewien problem.  
> Napisałem
> sobie taki oto przykładowy fragment kodu:
>
> my $queue = Thread::Queue->new();
> my $t1 = threads->create(\&test_reader);
> my $t2 = threads->create(\&test_writer);
>
> $t2->join();
>
> sub test_writer
> {
> 	while (1)
> 	{
> 		sleep 1;
> 		my %t = (1 => "A", 2 => "B")
> 		$queue->enqueue(share(%t));

Uwspólnianie zmiennych lokalnych najkrótszą drogą do wycieków pamięci.  
Owszem, w dokumentacji do Thread::Queue piszą o sharowaniu haszy przed  
kolejkowaniem w celu uniknięcia klonowania, ale dotyczy to zmiennych  
globalnych...

Tutaj jest problem, bo robienie takich zmiennych globalnych, to się samemu  
prosimy o problemy. Nie w tym testowym przypadku, ale jak więcej wątków ma  
pisać do tej kolejki, to bez mutexów będziemy sobie robić kuku.

Więc w zastosowaniach, gdzie prędkość nie jest sprawą kluczową pozwólmy,  
by Thread::Queue klonował sobie hasza.

W bardziej wymagających wydajnościowo zastosowaniach, lub przy bardziej  
złożonych strukturach danych przekazujmy tylko zwykłe skalary,  
serializując dane - jest sporo modułów do tego.

Natomiast to, co powodowało problem, to jest fakt, że Thread::Queue  
przekazuje tylko dane skalarne, czyli nie hasze, a referencje do nich...

Czyli pomijając zastrzeżenia wyżej, to powinno wyglądać raczej tak:

my %t : shared = (1 => "A", 2 => "B")
$queue->enqueue(\%t);

Jak używasz czegoś bardziej złożonego niż struktury proste, to użyj  
czegokolwiek  do serializacji, albo używaj Thread::Queue::Any.

> 	}	
> }
>
> sub test_reader
> {
> 	threads->detach();

Jak się odczepiasz(detach), to po co się potem przyczepiasz (join)? To nie  
zadziała i skrypt przestanie działać w momencie zakończenia się pierwszego  
wątku - drugi zostanie zabity mimo tego, że coś jest w kolejce.
	
> 	while (my $ref= $queue->dequeue())

Źle - wyjdzie Ci z pętli jak tylko kolejka stanie się pusta, w szczególnym  
przypadku, gdy drugi wątek zacznie działać przed tym, nim pierwszy skończy  
pierwszego sleepa, to w ogóle nic nie zrobi.

Lepiej pętlę utrzymywać na jakiejś kontrolowanej z zewnątrz zmiennej  
współdzielonej, a w niej zrobić drugą opartą na $queue->pending, a wyjadać  
z niej wersją nieblokującą (dequeue_nb)

> 	{
> 		my %hash = %{$ref}; # wyswietla np. HASH(0x1110970)
> 		print %hash; # tu juz nic nie wyswietla
> 	}
> }


Ja bym to zrobił jakoś tak:


#!/usr/bin/perl

use strict;
use threads;
use threads::shared;
use Thread::Queue;
use Data::Dumper;

my $queue = Thread::Queue->new();
my $QUIT : shared = 0;

my $t1 = threads->create(\&test_reader);
my $t2 = threads->create(\&test_writer);

for my $th (threads->list()) {
	$th->join;
}

sub test_writer {
	my $max = 10;
	while (!$QUIT) {
		sleep 1;
		
		my %t = (1 => "A", 2 => "B");
		$queue->enqueue(\%t);	
		if (--$max <= 0) {
			lock $QUIT;
			sleep 5; # by zdążyło wyjść z ew. sleepa z test_reader
			$QUIT = 1;
			
		}
		threads->yield();
	}	
}

sub test_reader {
	while(!$QUIT) {
		while ($queue->pending()){
			my $ref = $queue->dequeue_nb;
			print Dumper $ref;
		}
		threads->yield();
	}
}

Pozdrawiam
-- 
Vava
Wawrzyniec Żurowski
Victoria vale, et ubique es, suaviter sternutas

On Wed, 23 Dec 2009 21:31:04 +0100, Vava wrote:

>> 	while (my $ref= $queue->dequeue())
> 
> Źle - wyjdzie Ci z pętli jak tylko kolejka stanie się pusta, w szczególnym  
> przypadku, gdy drugi wątek zacznie działać przed tym, nim pierwszy skończy  
> pierwszego sleepa, to w ogóle nic nie zrobi.

Zaraz, dlaczego? To jest dequeue(), nie dequeue_nb(), które zwróciłoby undef
przy pustej kolejce; dequeue() powinno czekać, aż coś się w kolejce
znajdzie...


-- 
Szymon Sokół (SS316-RIPE) -- Network Manager                                 B
Computer Center, AGH - University of Science and Technology, Cracow, Poland  O
http://home.agh.edu.pl/szymon/ PGP key id: RSA: 0x2ABE016B, DSS: 0xF9289982  F
Free speech includes the right not to listen, if not interested -- Heinlein  H

Dnia 24-12-2009 o 00:15:06 Szymon Sokół  
<szymon@bastard.operator.from.hell.pl> napisał(a):

> On Wed, 23 Dec 2009 21:31:04 +0100, Vava wrote:
>
>>> 	while (my $ref= $queue->dequeue())
>>
>> Źle - wyjdzie Ci z pętli jak tylko kolejka stanie się pusta, w  
>> szczególnym
>> przypadku, gdy drugi wątek zacznie działać przed tym, nim pierwszy  
>> skończy
>> pierwszego sleepa, to w ogóle nic nie zrobi.
>
> Zaraz, dlaczego? To jest dequeue(), nie dequeue_nb(), które zwróciłoby  
> undef
> przy pustej kolejce; dequeue() powinno czekać, aż coś się w kolejce
> znajdzie...

Dobra, nie zauważyłem, że autor podłącza się do wątku piszącego. Jakoś  
wydawało mi się, że do czytającego. W takim razie nie wywali się przed  
tym, nim piszący coś zrobi, ale to zadziała wtedy, jeśli wątek czytający  
zdąży przetworzyć kolejkę nim piszący się zjoinuje (co w tym wyoadku  
zakończy program) - inaczej zostanie ubity.

Uproszczona wersja skryptu autora:

#!/usr/bin/perl

use strict;
use threads;
use threads::shared;
use Thread::Queue;

my $queue = Thread::Queue->new();
my $t1 = threads->create(\&test_reader);
my $t2 = threads->create(\&test_writer);

$t2->join;

sub test_writer {
	sleep 5;
	$queue->enqueue("TEST");	
}

sub test_reader {
	threads->detach();
	my $a = $queue->dequeue;
	#sleep 5;
	print $a;
}

Działa, acz przy obciążonym systemie nie musi.
Odkomentowanie sleepa w test_reader pokazuje, co się może stać.

W ogóle, to skrypt wątkotwórcy ma w tym miejscu jeszcze jedną wadę - pętla  
while (my $ref=$queue->dequeue) nie może się sama z siebie zakończyć...

Ja zdecydowanie wolę przy programowaniu wielowątkowych używać operacji  
nieblokujących. To wymaga trochę więcej myślenia, ale pozwala uniknąć  
sporej dawki problemów. W szczególności jeśli chcemy, by wątki ładnie  
pokończyły to co mają rozpoczęte, nawet po SIGTERM-ie.


Pozdrawiam
-- 
Vava
Wawrzyniec Żurowski
Victoria vale, et ubique es, suaviter sternutas

Dzięki za odpowiedzi, nieco się rozjaśniło. Jednak im dalej w las tym 
więcej drzew :) Kolejny problem z wątkami...

Otóż mam dwa programy w perlu: jeden używa bezpośrednio wątków a drugi 
Thread::Pool::Simple (kod funkcji wykonywanej przez wątek pozostaje 
niezmienny). Funkcja z grubsza wyglada tak:

sub foo
{
	my $bar = oblicz_coś();
	$out_queue->enqueue($bar);
}

I teraz dziwna rzecz: w programie w którym używam jednego wątku (threads-
>create()) kolejkę mam zadeklarowaną (w głównym programie, nie w funkcji) 
tak:

my $out_queue = Thread::Queue->new(); 

..i to działa. Jeśli jednak zamiast pojedynczego wątku używam Pool-a (z 
referencją metody "do" na tą samą funkcję co wcześniej) to kod wątku 
dodaje wyniki do kolejki ale w głównym programie kolejka pozostaje pusta 
(po zjoinowaniu Poola) :/ Pomaga dopiero zmiana deklaracji kolejki na:

our $out_queue : shared = Threads::Queue->new();

Dlaczego tak się dzieje? Przyznam, że nieco się zamotałem i chciałbym 
nieco uporządkować wiedzę. Nie rozumiem dlaczego *ten sam* kod wątku 
działa różnie w zależności od tego czy używam Poola czy nie.

-- 
pozdrawiam,
menth0l

Dnia 27-12-2009 o 19:31:24 Krzysztof Lewicki <menth0l@remove.this.o2.pl>  
napisał(a):

> Dzięki za odpowiedzi, nieco się rozjaśniło. Jednak im dalej w las tym
> więcej drzew :) Kolejny problem z wątkami...
>
> Otóż mam dwa programy w perlu: jeden używa bezpośrednio wątków a drugi
> Thread::Pool::Simple (kod funkcji wykonywanej przez wątek pozostaje
> niezmienny). Funkcja z grubsza wyglada tak:
>
> sub foo
> {
> 	my $bar = oblicz_coś();
> 	$out_queue->enqueue($bar);
> }
>
> I teraz dziwna rzecz: w programie w którym używam jednego wątku (threads-
>> create()) kolejkę mam zadeklarowaną (w głównym programie, nie w funkcji)
> tak:
>
> my $out_queue = Thread::Queue->new();
>
> ..i to działa. Jeśli jednak zamiast pojedynczego wątku używam Pool-a (z
> referencją metody "do" na tą samą funkcję co wcześniej) to kod wątku
> dodaje wyniki do kolejki ale w głównym programie kolejka pozostaje pusta
> (po zjoinowaniu Poola) :/ Pomaga dopiero zmiana deklaracji kolejki na:
>
> our $out_queue : shared = Threads::Queue->new();
>
> Dlaczego tak się dzieje? Przyznam, że nieco się zamotałem i chciałbym
> nieco uporządkować wiedzę. Nie rozumiem dlaczego *ten sam* kod wątku
> działa różnie w zależności od tego czy używam Poola czy nie.


Nie bawiłem się nigdy Thread::Pool::Simple, ale taki kod:


#!/usr/bin/perl

use strict;
use threads;
use threads::shared;
use Thread::Queue;
use Thread::Pool::Simple;

my $QUIT : shared = 0;
my $out_queue = Thread::Queue->new();

my $l = threads->create('listener');

my $pool = Thread::Pool::Simple->new(
	min => 3,
         max => 5,	
	do => [\&foo,],
	passid => 1,
);

$out_queue->enqueue("\t...TH...");
my @a;
push @a, threads->create('foo', $_) for 1..10;
$_->join for @a;

$out_queue->enqueue("\t...POOL...");


$pool->add(1) for (1..10);
$pool->join;

lock $QUIT;
$QUIT=1;
$l->join;


sub foo {
	my $bar = oblicz_cos();
	$out_queue->enqueue($_[0] . ": " . $bar);
}

sub oblicz_cos {
	return int(rand(100))+1;
}

sub listener {
	while (!$QUIT) {
		if ($out_queue) {
			while (my $msg = $out_queue->dequeue_nb) {
				print "$msg\n";
			}
		}
	}
}


__END__


Wydaje się działać:

[vava@valis: /home/vava/test]$ ./test.pl

         ...TH...
1: 30
2: 74
3: 54
4: 80
5: 43
6: 38
7: 82
9: 23
8: 50
10: 27
         ...POOL...
975321554: 63
587457719: 10
884075336: 13
305938673: 20
482268431: 75
479418188: 47
190195793: 62
466707656: 54
184764929: 8
1171988162: 67

Pozdrawiam
-- 
Vava
Wawrzyniec Żurowski
Victoria vale, et ubique es, suaviter sternutas