Kolejka w PHP, Opcje

[Prph]

Dawno mnie tu nie było

Forumowicze, problem z kolejką w PHP. System, który tworzę przetwarza dane co minutę (z CRONa). Niestety uruchomienie pojedynczego procesu np, przetwarzaj.php nie kończy się w 1 minucie, ponieważ przetwarzanie danych może trwać np. 2 mninuty dla pojedynczego rekordu bazy danych, a rekordów jest np. 20. Teoretycznie szeregowe uruchomienie przetwarzania zajmie więc 40 min. Każdy przetwarzaj.pho bierze z bazy zatem 1 rekord i tylko taki mieli.

Moje pytanie: w jaki sposób napisać kolejkę, która poprawnie przetworzy takie dane? Wywołać z crona 20x ten sam przetwarzaj.php? Zanim proces zacznie wykonywać swoją pracę, ustawia na rekordzie bazy, że jest on przetwarzany, więc inny proces weźmie kolejne dane.

Spotkaliście się z podobnym problemem, macie pomysły?

[Crozin]

Podstawowe pytanie: czy przetworzenie kilku, np. maksymalnie pięciu, elementów na raz działa szybciej od przetwarzania kolejnych elementów po kolei?

[Prph]

Tak, przecież zostaną wykonane równolegle. Wykonają się zatem w ciągu 2 minut, zamiast 40.

[!*!]

Tak, przecież zostaną wykonane równolegle. Wykonają się zatem w ciągu 2 minut, zamiast 40.

Chciałbyś. Przecież "moc" obliczeń jest identyczna,więc jak zaginasz czasoprzestrzeń?

Ten post edytował !*! 23.05.2012, 07:13:39

[hind]

Zły tok rozumowania... równoległe uruchomienie spwoduje że wszystkie procesy będą działać wolniej... więc zamiast upragnionych 2 minut będzie to 20 (lub i więcej)...

jak zrobić?
napisać skrypt który w pętli odpali te 20 zadań np

[PHP] pobierz, plaintext 
#!/usr/bin/env php
<?php
foreach(range(1,20) as $v) {
echo `/skrypt/do/eykonania.php --id=$v &`; // & <- b. ważny znak
}
[PHP] pobierz, plaintext 

Ten post edytował hind 23.05.2012, 07:05:43

[Prph]

No bez przesady. 16 rdzeni przetworzy szybciej 16 równoległych procesów, niż 1 ciągły. Proces nie może wykonać się szeregowo, bo już pierwsze przetwarzanie danych może zająć dużo czasu, a przetwarzanie musi odbywać się co minutę. Pomysł hind wnosi coś do dyskusji.

[Crozin]

To może inaczej, co jest powodem tak długiego przetwarzania pojedynczego rekordu? Bo jeżeli są to jakieś ciężkie operacje wykorzystujące procesor albo dysk, wykonanie równoległych przetwarzań wcale nie musi przyspieszyć ich ogólnego czasu wykonywania. Zaś jeżeli jest to spowodowane przykładowo oczekiwaniem na pobranie danych z sieci to już jak najbardziej miałoby to sens.

1. Zamiast odpalać co minutę zadanie w Cronie, utwórz sobie daemona(y) (System_Daemon).
2. Generalnie utworzenie w bazie danych kolumny status (awaiting, processing, processed) będzie niewystarczające. Dwa (lub nawet więcej) procesy mogłyby pobrać ten sam wiersz nim którykolwiek z nich zdąży zmienić jego status z awaiting na processing. Tutaj trzeba by było albo zablokować na chwilę tabelę dla odczytu:

[SQL] pobierz, plaintext 
LOCK TABLE tbl_name READ;
SELECT ... WHERE STATUS = 'awaiting' LIMIT 1;
UPDATE ... SET STATUS = 'processing' WHERE id = ...;
UNLOCK TABLE tbl_name;
[SQL] pobierz, plaintext 

3. Jeżeli masz możliwość nie pisz tego w PHP, tylko w czymś co wspiera współbieżność (ot, chociażby Java) - znacznie uprości Ci to życie.

EDIT:
Ewentualnie zamiast ręcznego bawienia się w blokowanie tabeli skorzystaj z transakcji, i tak zawsze trzeba, na odpowiednim poziomie izolacji - tutaj potrzebny byłby chyba aż poziom serializable.

Ten post edytował Crozin 23.05.2012, 07:38:35

[Prph]

No to jest odpowiedź na wysokim poziomie

Proces nie obciąża procesora. Długo wykonuje się, ponieważ przesyła dane i też niedużo, ale klient odbierający dane przetwarza je i to zajmuje czas.

Nawet gdyby 2 procesy dobrały się do tych samych danych nic się nie stanie, klient otrzyma je drugi raz co będzie skutkowało najwyżej ponownym przemieleniem danych.

Jeśli chodzi o Javę, to faktycznie mógłbym jej użyć, ale to banalny kod w PHP, jeśli udałoby się tylko wywołać 20 równoległych procesów to myślę, że byłoby OK.

Co z forkowaniem? Jak ten Demon w PHP? Ktoś to wdrażał w profesjonalnych zastosowaniach?

[Crozin]

Proces nie obciąża procesora. Długo wykonuje się, ponieważ przesyła dane i też niedużo, ale klient odbierający dane przetwarza je i to zajmuje czas.

Tak więc tutaj może to mieć rzeczywiście sens. Musisz tylko pamiętać, że jednoczesne podesłanie znaczącej ilości żądań do drugiej strony może ją "zamulić" i w efekcie jeszcze bardziej wydłużyć ogólny czas działania. Prawdopodobnie dobrym pomysłem będzie wysyłanie jakiejś ograniczonej ilości żądań jednocześnie, np. 5 albo 20.

Jeśli chodzi o Javę, to faktycznie mógłbym jej użyć, ale to banalny kod w PHP, jeśli udałoby się tylko wywołać 20 równoległych procesów to myślę, że byłoby OK.

Tak na pierwszy rzut oka, wykonanie tego np. w Javie będzie znacznie prostsze niż w PHP. Jeden wątek pobierający rekordy do obrobienia (w ilości np. 100), który będzie je wrzucać na stos oraz 5-10-20 wątków pobierających po jednym elemencie ze stosu i przetwarzających go. W momencie gdy ilość rekordów na stosie się kończy będzie on pobierać sobie następną porcję.
Zero problemów z wielokrotnym obrabianiem tego samego rekordu, "prosta" baza danych nie wymagająca żadnego dbania o współbieżność (brak LOCKów), jeden proces w systemie zamiast n+1 (czyli łatwe okiełznanie tego). Całość pewnie w 50 linijkach kodu by się zmieściła, z czego połowa to pewnie by były klamerki.

Co z forkowaniem? Jak ten Demon w PHP? Ktoś to wdrażał w profesjonalnych zastosowaniach?

W forkowanie nigdy się w PHP nie bawiłem - ono się po prostu do tego kompletnie nie nadaje. Daemon w PHP? Jak najbardziej, ale pojedynczy.

EDIT: Oczywiście strukturą danych przechowującą rekordy do przetworzenia powinna być tytułowa kolejka, nie stos.

Ten post edytował Crozin 23.05.2012, 19:43:37

[Niktoś]

Php z natury nie jest wielowątkowe(czyt.threading), pozostaje kolejkowanie, albo "emulacja" wielowątkowości poprzez użycie pamięci wspóldzielonej.
Taka implementacja wielowątkowości w PHP:
http://www.php.blogle.info/2011/nietypowe-...atkowosc-w-php/

Ten post edytował Niktoś 23.05.2012, 19:16:07

[Prph]

Zrobione za pomocą forkowania. Aż miło patrzeć jak to działa

[PHP] pobierz, plaintext 
 
	$sql ='tutaj selekt pobierający listę do wysłania...';
 
	$res = db_query($sql);
	$queue = db_fetch_all($res);
 
	if(!count($queue)) {
		echo "Nothing to send\n";
	}
 
	$numOfPids = 0;
	$processes = array();
	foreach($queue as $entry) {
 
		$processes[$numOfPids] = pcntl_fork();
		if($processes[$numOfPids]) { // this is parent process
			$numOfPids++;
		}
		else {
 
			echo "In $numOfPids process, sending {$entry['id']}\n";
 
			// ok we are in the child
			// send data
 
			// have to connect again
			db_connect('...');
 
			$update = array('status' => 'in_process');
			db_update('...'));
			send_data($entry); // funkcja wysyłająca
 
			echo "Process $numOfPids sent data\n";
 
			exit();
		}
	}
[PHP] pobierz, plaintext 

Wynik:

[PHP] pobierz, plaintext 
In 0 process, sending 5790
In 1 process, sending 5791
In 2 process, sending 5797
Process 0 sent data
Process 2 sent data
Process 1 sent data
It's done!
[PHP] pobierz, plaintext 

[Crozin]

Aż mnie coś wzięło, żeby zobaczyć jak można zrobić to "normalnie" (czyt: bez forkowania procesów, a z użyciem normalnych wątków):

[JAVA] pobierz, plaintext 
package com.crozin.concurrent;
 
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
 
public class RecordsProcessor implements Runnable {
	private static final int NUMBER_OF_PROCESSORS = 8;
 
	private final BlockingQueue<Integer> processingQueue = new LinkedBlockingQueue<Integer>();
 
	private class ProcessorThread extends Thread {
		public ProcessorThread(String name) {
			super(name);
		}
 
		@Override
		public void run() {
			try {
				while (true) {
					Integer record = processingQueue.take();
 
					log("Processing record #" + record);
 
					Thread.sleep(new java.util.Random().nextInt(1000) + 1000);
				}
			} catch (InterruptedException e) {
				return;
			}
		}
	}
 
	@Override
	public void run() {
		for (int i = 0; i < NUMBER_OF_PROCESSORS; i++) {
			new ProcessorThread("Processor #" + i).start();
		}
 
		int c = 1;
 
		while (true) {
			if (processingQueue.size() <= 10) {
				log("Refilling the queue");
 
				for (int i = 0; i < 100; i++) {
					processingQueue.add(c++);
				}
			}
 
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				return;
			}
		}
	}
 
	public static void main(String[] args) throws Exception {
		new RecordsProcessor().run();
	}
 
	public static void log(String msg) {
		System.out.println(Thread.currentThread().getName() + ": " + msg);
	}
}
[JAVA] pobierz, plaintext 

Minusy:
- konieczność zrezygnowania z PHP dla tego elementu systemu.

Plusy:
- będzie działać na każdej platformie,
- gwarantuje, że rekordy zostaną przetworzone w odpowiedniej kolejności oraz tylko i wyłącznie jeden raz,
- brak niepotrzebnych przerw w działaniu,
- kontrola nad ilością przetwarzanych rekordów w danym momencie,
- łatwiej to wszystko ogarnąć, wprowadzać zmiany itp.

[cezet]

Witajcie,

Pewnie mnie teraz zjedziecie, bo jak śmiem...
Ale zaryzykuję. Uważam, że metody które podaliście, to zabijanie muchy przy użyciu czołgu.

Ja swojego czasu miałem podobny problem. Potrzebowałem mechanizmu częstego uruchamiania procesu, który może trwać chwilę, lub pół godziny, i warunek - procesy nie mogły działać równolegle.

I rozwiązałem to na zasadznie sprawdzania pliku typu lock.
czyli coś w stylu:

jeden plik odpalany np. co minutę z crona (np cron.php)

[PHP] pobierz, plaintext 
<?php
if(!file_exists('lock.lck')) {
if(touch('lock.lck')) {
include('zrob_cos.php');
unlink('lock.lck');
}
}
?>
[PHP] pobierz, plaintext 

oraz drugi plik zrob_cos.php który wykonywał nasze zadania.

To taki ogólny zarys, ale.. proste, banalne i skuteczne

Ten post edytował cezet 30.05.2012, 13:13:31

[Crozin]

[...] procesy nie mogły działać równolegle.

A zauważyłeś, że cały wątek tyczy się właśnie równoległego odpalenia wielu procesów/wątków?

[greycoffey]

Nie zjedziemy, acz wytkniemy błąd.
Ten skrypt to tylko semafor zapobiegający uruchomieniu dwóch instancji tego skryptu. Autor natomiast chciałby równloegle przetwarzać te rekordy, rozłożyć je na kilka wątków (serwer jak autor pisał 16-rdzeniowy), dzięki czemu zadania będą się wykonywały równolegle.

[cezet]

Zwracam honor