Reads lines of work from an input stream and farms them out to multiple child streams. More...

Inheritance diagram for OrderedStreamingForkController:

Collaboration diagram for OrderedStreamingForkController:

[legend]

Public Member Functions
	__construct ( $numProcs, $workCallback, $input, $output)

	start ()
	@inheritDoc

Public Member Functions inherited from ForkController
	__construct ( $numProcs, $flags=0)

	getChildNumber ()
	Get the number of the child currently running.

Protected Member Functions
	consume ()
	Child worker process.

	consumeNoFork ()
	Special cased version of self::consume() when no forking occurs.

	feedChildren (array $sockets)
	Reads lines of work from $this->input and farms them out to the provided socket.

	forkWorkers ( $numProcs)

	receive ( $id, $data)

	updateAvailableSockets (&$sockets, &$used, $timeout)
	Moves sockets from $used to $sockets when they are available for more work.

Protected Member Functions inherited from ForkController
	handleTermSignal ( $signal)

	initChild ()

	prepareEnvironment ()

Protected Attributes
string[]	$delayedOutputData = []
	Int key indicates order, value is data.

resource	$input

int	$nextOutputId

resource	$output

callable	$workCallback

Protected Attributes inherited from ForkController
	$childNumber = 0

	$children = []

	$flags = 0

	$procsToStart = 0

	$termReceived = false

Additional Inherited Members
Public Attributes inherited from ForkController
const	RESTART_ON_ERROR = 1
	Pass this flag to __construct() to cause the class to automatically restart workers that exit with non-zero exit status or a signal such as SIGSEGV.

Static Protected Attributes inherited from ForkController
static	$restartableSignals

Detailed Description

Reads lines of work from an input stream and farms them out to multiple child streams.

Each child has exactly one piece of work in flight at a given moment. Writes the result of child's work to an output stream. If numProcs <= zero the work will be performed in process.

This class amends ForkController with the requirement that the output is produced in the same exact order as input values were.

Currently used by CirrusSearch extension to implement CLI search script.

Since: 1.30

This program is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation; either version 2 of the License, or (at your option) any later version.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with this program; if not, write to the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. http://www.gnu.org/copyleft/gpl.html

Definition at line 32 of file OrderedStreamingForkController.php.

Constructor & Destructor Documentation

◆ __construct()

OrderedStreamingForkController::__construct	(	$numProcs,
		$workCallback,
		$input,
		$output )

Parameters

int	$numProcs	The number of worker processes to fork
callable	$workCallback	A callback to call in the child process once for each line of work to process.
resource	$input	A socket to read work lines from
resource	$output	A socket to write the result of work to.

Definition at line 51 of file OrderedStreamingForkController.php.

References $input, $output, and $workCallback.

Member Function Documentation

◆ consume()

OrderedStreamingForkController::consume ( )

protected

Child worker process.

Reads work from $this->input and writes the result of that work to $this->output when completed.

Definition at line 121 of file OrderedStreamingForkController.php.

References $data, $line, and(), and list.

Referenced by start().

◆ consumeNoFork()

OrderedStreamingForkController::consumeNoFork ( )

protected

Special cased version of self::consume() when no forking occurs.

Definition at line 135 of file OrderedStreamingForkController.php.

References $data, and and().

Referenced by start().

◆ feedChildren()

OrderedStreamingForkController::feedChildren ( array $sockets )

protected

Reads lines of work from $this->input and farms them out to the provided socket.

Parameters

resource[] $sockets

Definition at line 157 of file OrderedStreamingForkController.php.

References $data, and(), and updateAvailableSockets().

Referenced by forkWorkers().

◆ forkWorkers()

OrderedStreamingForkController::forkWorkers ( $numProcs )

protected

Parameters

int $numProcs

Returns: string

Reimplemented from ForkController.

Definition at line 78 of file OrderedStreamingForkController.php.

References and(), feedChildren(), ForkController\initChild(), and ForkController\prepareEnvironment().

◆ receive()

OrderedStreamingForkController::receive	(		$id,
			$data )

protected

Parameters

int	$id
string	$data

Definition at line 213 of file OrderedStreamingForkController.php.

References $data, and and().

Referenced by updateAvailableSockets().

◆ start()

OrderedStreamingForkController::start ( )

@inheritDoc

Reimplemented from ForkController.

Definition at line 61 of file OrderedStreamingForkController.php.

References consume(), and consumeNoFork().

◆ updateAvailableSockets()

OrderedStreamingForkController::updateAvailableSockets	(	&	$sockets,
		&	$used,
			$timeout )

protected

Moves sockets from $used to $sockets when they are available for more work.

Parameters

resource[]	&$sockets	List of sockets that are waiting for work
resource[]	&$used	List of sockets currently performing work
int	$timeout	The number of seconds to block waiting. 0 for non-blocking operation.

Definition at line 195 of file OrderedStreamingForkController.php.

References $data, $line, and(), list, and receive().

Referenced by feedChildren().