How to Thread a class method

Question

I need to perform a parallel process using php, I installed the extension pthreads and I looked at some examples but didn’t see how I could implement it.

Class:

<?php

class Negociacao{
    /*
        Retorna código de rastreio das condições do usuário.
    */
    public function SolicitarOpcoesPagamento($cpfOuCnpj, $telefone, $email, $nrParcelas, $dataPagamento){
        $codigo =  md5($cpfOuCnpj); 
        //ConsultarOpcoesPagamento($codigo); <- aqui dou start na thread
        return $codigo;
    } 

    /*
        Retorna status e array de condições do usuário.
    */
    public function ConsultarOpcoesPagamento($codigo){
        return array("status" => "pendente","opcoes"=> array());
    }

}

Soap server:

<?php

$options = array("uri" => "http://localhost/soap/");

$server = new SoapServer(null, $options);

require_once "negociacao.php";

$server->setObject(new Negociacao());

$server->handle();

?>

Cliente Soap:

<?php
$options = array(
    'uri' => 'http://localhost/soap/server.php',
    'location' => 'http://localhost/soap/server.php'
);

$client = new SoapClient(null, $options);

var_dump($client->solicitarOpcoesPagamento(025111151544));

?>

Goal:

When the customer calls the function SolicitarOpcoesPagamento i must return a tracking id immediately, but before returning this id I need to run inside this function in parallel to the function ConsultarOpcoesPagamento, that will go on another webservice, assemble the options and insert in my database. When the customer wants it will query the function ConsultarOpcoesPagamento that will already be with the data.

Answer 1

I’m not sure I quite understand your problem, but if I do, I believe you won’t need to create a thread just for that. What you can do is deliver the HTTP response of the request to the client and continue with your code. In the example below I adapted some of your code just for testing.

class Negociacao {

    public function SolicitarOpcoesPagamento($data){
        $codigo =  md5($data); 
        return $codigo;
    } 

    public function ConsultarOpcoesPagamento($codigo){
        file_put_contents(
            "output.json", 
            json_encode(
                array(
                    "status" => "pendente",
                    "opcoes"=> array(
                        "codigo" => $codigo
                    )
                )
            )
        );
    }

}

That is, the method ConsultarOpcoesPagamento create a unique code from any data and the method ConsultarOpcoesPagamento will generate a basic JSON containing the information generated by the previous method. In your case, this method would work with the database as per your real need. To finalize the request and send the response to the customer, you can do something like:

<?php

ignore_user_abort(true);
set_time_limit(0);

ob_start();

$neg = new Negociacao();
$codigo = $neg->SolicitarOpcoesPagamento("Foo");

echo "Seu código de acesso é: ", $codigo;

header('Connection: close');
header('Content-Length: '.ob_get_length());

ob_end_flush();
ob_flush();
flush();

// Seu código continuará aqui...

In this case, when opening the PHP page, the client will get the following message:

Seu código de acesso é: 1356c67d7ad1638d816bfb822dd2c25d

But the code will continue running. To simulate a longer process, I will use the function sleep and after invoking the method ConsultarOpcoesPagamento:

<?php

ignore_user_abort(true);
set_time_limit(0);

ob_start();

$neg = new Negociacao();
$codigo = $neg->SolicitarOpcoesPagamento("Foo");

echo "Seu código de acesso é: ", $codigo;

header('Connection: close');
header('Content-Length: '.ob_get_length());

ob_end_flush();
ob_flush();
flush();

// Seu código continuará aqui...

sleep(5);

$neg->ConsultarOpcoesPagamento($codigo);

So, 5 seconds after you get the answer in the browser, the file output.json will be created with the content:

{"status":"pendente","opcoes":{"codigo":"1356c67d7ad1638d816bfb822dd2c25d"}}

Answer 2

Maybe your question isn’t about threads themselves, but this question will be accessed by people looking for solutions about threads in php.

First of all, it is important to understand the definition of what a threads is:

In computer science, Thread is a small program that works as a subsystem, being a way of a process self-divide into two or more tasks.

This is a very complex procedure to perform with PHP, since parallelism is not usually the focus of PHP programmers, but it is not impossible to do. And there are still several approaches, some simpler and safer others not so much. See below for some methods to perform or simulate threads in php.

Method 1: Create a call to your own server

The first, simpler approach is defined in this answer: /a/217559/7130 However this method is not an asynchronous execution and would be less performative than calling a method directly within the same process.

Method 2: Cronjobs

Another approach, also very simple is cron or cronjob, in windows task scheduler. The technique consists of creating a routine that is read from time to time behind some "task" that is scheduled, you can store this list of tasks to be performed by the script in a database or file with the parameters. It’s not very performative either, but it’s simpler and easier to implement, as long as you have access to a task scheduling feature.

Method 3: PHP Process Control (PCNTL )

Well, all of the above methods cannot be executed asynchronously, separating the php process into 2. Maybe this is the only way to thread a class method. Needing to create a solution that did just that, where it was possible to initialize a routine/method that ran in the background while I already gave the return to the user and then just checked the progress and log of this task.

The result was the elaboration of an abstract class to be extended by sub-classes with specific tasks, in the example below in the gist link I do the deletion of files in asynchronous form and record logs during the process See part of the method below:

/**
     * Executa a tarefa com os parâmetros pré-definidos.
     * @param mixed $parameters
     */
    final public function run($parameters) {
        try {
            //Não tenho certeza se há necessidade, 
            //porém essas funções adicionam um controle para que o 
            //processo continue à ser execudado mesmo que o usuário abandone a tela.
            ignore_user_abort(true);
            set_time_limit(0);
            // Inicializo o signal Handler
            pcntl_signal(SIGCHLD, SIG_IGN);
            $pid = pcntl_fork(); //Cria um novo processo, se criado com sucesso, o código executado daqui para baixo será executado em outra thread.
            if ($pid == -1) {//Se o retorno for -1 significa que não foi possível.
                throw new \RuntimeException("Could not start a fork process.");
            } else if ($pid) {
                //Se o valor for diferente de zero Significa que ainda está rodando no pai
//                pcntl_waitpid($pid, $exitcode, WNOHANG); // prevents zombie processes
//                 $this->log("Thread Still the same");
                return;
            } else {
                ignore_user_abort();
                $sid = posix_setsid();
                self::$shmId = shm_attach((int) (ftok(self::$file, 'A') . self::$line), self::SHARED_MEMORY_SIZE);
                $this->var_key_fork_pid = $this->alocatesharedMemory($sid, 112105100); //112105100
                shm_put_var(self::$shmId, $this->var_key_status, 'RUNNING');
//                $this->log("SubTask" . $sid);
//                $this->log("SubPid" . $this->var_key_fork_pid);
//                $this->log("Substatus" . $this->var_key_status);
                $this->onPreExecute($parameters);
//                $this->log("SessionDelay: " . session_id());
                try {
                    $result = $this->doInBackground($parameters);
                } catch (\Exception $e) {
                    $this->log("[ErrExec:" . $e->getCode() . "] " . $e->getMessage());
                    $this->cancel($e->getMessage());
                }
                $this->onPostExecute($result);
                //Ao finalizar, emitir o sinal de finalizado e apagar a memória compartilhada alocada.
                if (@shm_has_var(self::$shmId, $this->var_key_fork_pid)) {
                    shm_put_var(self::$shmId, $this->var_key_fork_pid, null);
                    shm_put_var(self::$shmId, $this->var_key_status, 'FINISHED');
                    shm_remove(self::$shmId);
                    gc_collect_cycles();
                    posix_kill(getmypid(), SIGKILL);
                }
            }
        } catch (\Exception $e) {
            $msg = "[Err:" . $e->getCode() . "] " . $e->getMessage(); 
            $this->log($msg);
            self::onError($msg,$e);
        }
    }

Explanation: Thread separation happens on the line where I use the following function: $pid = pcntl_fork(); this function can return -1 if it fails to create the process branch (Fork). The value other than 0 is the block that will continue running in the main process the other value is the process id "child" or Fork, this block will run in a completely separate process. To make the communication between the 2 I use the shared memory, since they are 2 separate processes and both do not have access to the same variables (not even for $_SESSION).

So inside your example could be something like this:

if($pid == 0){ //Se estiver na thread principal
 SolicitarOpcoesPagamento();
}else if($pid != -1){ //Se tiver no fork
 ConsultarOpcoesPagamento();
}

Complete Example: https://gist.github.com/LeonanCarvalho/62c6fe0b62db8a478f502f84c5734c83 In my example you would need to model a class for each task, these classes should inherit from the abstract class and implement some methods.

It consists of using the pcntl functions, which is the official php extension for process control, mainly the function pcntl_fork to create a separation from the main process. It was also necessary to use some auxiliary extensions such as the POSIX and "Semaphore, Shared Memory and IPC".

Result of the above example:

One problem known to me is that it is not possible to use PDO instances in Singleton mode between 2 processes, you need in the branched process ("Fork") destroy the Singleton instances of PDO and then create a new PDO connection.

This method is in use for 1 year, it is responsible for importing files into the system, instead of occupying the connection with the server I just receive the data and then process it in the background on the server, freeing the user connection so that it continues to work on the system.

Method 4: Pthreads

Although it already exists in php 5.6, I strongly recommend using it only in PHP7, it has been completely refactored in PHP7 and is safer to use. This will require a ZTS (Zend Thread Safety) version of PHP 7.x installed, along with pthreads v3 installed. So far I have not found link to download this version ready, but you can still compile it https://www.sitepoint.com/install-php-extensions-source/#Installing-a-third-party-Extension ( Link in English).

The official documentation is: http://php.net/manual/en/book.pthreads.php

But the good examples are here: https://www.sitepoint.com/parallel-programming-pthreads-php-fundamentals/ (link in English)

In the specific case of the question, I do not know if these last 2 methods would meet, there is no way you can be sure that the ConsultarOpcoesPagamento will end in parallel with the ConsultarOpcoesPagamento, but it would look something like this:

class OpcoesPagamento extends Threaded
{
    public $codigo;
    public $opcoes;

    public function Solicitar($cpfOuCnpj, $telefone, $email, $nrParcelas, $dataPagamento)
    {
        /** **/
        $this->codigo = md5($cpfOuCnpj); 
    }

    public function Consultar()
    {
        $this->opcoes = array("status" => "pendente","opcoes"=> array());
    }
}
$OpcoesPagamento = new OpcoesPagamento;


$thread = new class($OpcoesPagamento) extends Thread {
    private $task;

    public function __construct(Threaded $task)
    {
        $this->task = $task;
    }

    public function run()
    {
        $this->task->Solicitar();
        $this->task->Consultar();       
    }
};
$thread->start() && $thread->join();

var_dump($task->codigo);
var_dump($task->opcoes);

Another relevant project that may be of interest is this one with plenty of examples: https://github.com/krakjoe/pthreads, also requires PHP7. I couldn’t find a linux compatible version.

In short, there are several ways to perform a Thread in a class method, some simulating and others with specific components, either asynchronous or synchronous. Which you use should be based on your requirements and your infrastructure, since they need a little advanced knowledge to implement any of these techniques.