Download the PHP package wapmorgan/threadable without Composer
On this page you can find all versions of the php package wapmorgan/threadable. It is possible to download/install these versions without Composer. Possible dependencies are resolved automatically.
Informations about the package threadable
Threadable
Easy-to-use threading library providing all basic features to perform work in background mode.
All you need to have installed:
- pcntl
- posix
- sockets
This library can also work in simulation mode, where no actual forking performs. All work is done in one main thread.
This mode enables if pnctl extension is not available or when you specify it in Worker
constructor.
- Structure
- What is a
Worker
?- How to create your Worker
- What is a
WorkersPool
?
- What is a
- Simple usage
- How it works
- One worker
- Few workers with
WorkersPool
- API
Worker
APIWorkersPool
API
- Predefined workers
DownloadWorker
- Use cases
Structure
What is a Worker?
Worker - is a basic class for any worker. It is composed of two substances (physically, stored in one class, but providing different functionalities):
- A
Worker
- a separate thread, doing all background work. - A
Worker
manager - a manipulator for the worker thread.
How to create your Worker
The all you need it to extend wapmorgan\Threadable\Worker
class and reimplement onPayload($data)
public method.
For example:
What is a WorkersPool?
WorkersPool (wapmorgan\Threadable\WorkersPool) - is a container for Worker
's, intended for handling similar tasks.
It takes care of all maintenance, payload dispatching and life-cycle of workers. Allows you change the size of the pool dynamically and other useful stuff.
Simple usage
For example, you want to just background downloading work. Let's use wapmorgan\Threadable\BackgroundWork
class to background it and show progress for user (or store in DB/...).
Everything you need to do:
- Prepare payloads for
DownloadWorker
- Launch
BackgroundWork::doInBackground()
orBackgroundWork::doInBackgroundParallel()
for one thread or few threads respectively.
Stage 1. Preparing payloads
DownloadWorker
needs an array with source
and target
elements. Prepare it:
Stage 2. Launching in background
One-thread worker
Run it in one thread with doInBackground
function. Signature is following:
doInBackground(Worker $worker, array $payloads, callable $payloadHandlingCallback = null, callable $onPayloadFinishCallback = null, $sleepMicroTime = 1000)
$worker
- an instance of worker.$payloads
- an array of all payloads.$payloadHandlingCallback
- a callback that will be called every$sleepMicrotime
microseconds with information about currently running payload. Signature for callback:(Worker $worker, int $payloadI, $payloadData)
$onPayloadFinishCallback
- a callback that will be called when worker ends with one payload. Signature for callback:(Worker $worker, int $payloadI, $payloadData, $payloadResult)
So, collect all information to run it:
Example is in bin/example_file_downloading_easy
file.
Few-threads worker
To run it in few threads use doInBackgroundParallel
. It has almost the same signature as one-thread function:
doInBackgroundParallel(Worker $worker, array $payloads, callable $payloadHandlingCallback = null, callable $onPayloadFinishCallback = null, $sleepMicroTime = 1000, $poolSize = self::BY_CPU_NUMBER)
By adjusting $poolSize
you can select number of workers that should be used.
Example is in bin/example_file_downloading_pool_easy
file.
How it works
One worker
If you just need to parallel some work and do it in another thread, you can utilize just Worker
class without any other dependencies.
To use it correctly you need to understand the life-cycle of worker:
- Worker starts in another thread. To do this call
start()
. - Worker accepts new payload and starts working on it. To do this call
sendPayload(array $data)
. Really, worker manager sends payload via local socket. Worker thread starts working on it and returns result of work on finish via the same socket. - Worker manager checks if worker thread has done and read result of work. To do this call
checkForFinish()
. - Worker stops or being killed by
stop()
orkill()
methods respectively. - Worker manager checks if worker thread has finished and marks itself terminated. To do this call
checkForTermination()
.
Background work happens in 2 steps, where worker thread runs onPayload($data)
method of class with actual payload.
To summarize, this is an example of downloading file in another thread with real-time displaying of progress:
Settings and structures
Real work
Result:
This code equipped with a lot of comments, but you can simplify this example if you don't need to re-use worker when all your work is done. You can replace this huge loop with a smaller one:
Result:
Few workers with WorkersPool
But what if you need do few jobs simultaneously? You can create few instances of your worker, but it will be pain in the a$$ to manipulate and synchronize them.
In this case you can use WorkersPool
, which takes care of following this:
- Start new workers at the beginning.
- Dispatch your payload when you call sendData to any idle worker*.
- Create new workers or delete redundant workers when you change poolSize.
- Accept result of workers when they done and marks them as idle.
- Monitor all worker threads and count idle, running, active (idle or running) workers. Provides interfaces to acquire this information.
- Stop all workers when
WorkersPool
object is being destructed (viaunset()
or when script execution is going down). - *Can work in dataOverhead-mode. This mode enables sending extra payload to workers even when are already working on any task. If in this mode you sent few payloads to worker, it will not switch to Worker::IDLE state until all passed payloads have been processed.
- Provide interface to appoint progress trackers and run them periodically until all threads become in
Worker::IDLE
state.
Rich feature-set, right?! Let's rewrite our downloader with 2 threads to speed-up downloading.
The Settings and structures block of code remains the same, but for demonstating purposes let's use two big files:
We need to update only code working with threads.
Result:
As you can see, we got few improvements:
- Our code became smaller and clearer.
- We can run as many workers as we need.
- We don't take care of worker termination anymore. Let WorkersPool work for us.
API
Worker API
sendPayload($data): int
- sends payload to worker and returns serial id for payload.checkForFinish(): array|null
- checks if worker sent result of payload and returns it in this case.checkForTermination(): boolean|null
- returns true if worker process has died.stop($wait = false): boolean
- sends stop command to worker thread. It uses SIGTERM signal to allow worker thread finish work correctly and don't lose any data. If$wait = true
, holds the execution until the worker is down.kill($wait = false): boolean
- sends stop command to worker thread. It uses SIGKILL signal and not recommended except special cases, because it simply kills the worker thread and it loses all data being processed in that moment. If$wait = true
, holds the execution until the worker is down.
Information:
isActive(): boolean
- true if worker is inWorker::RUNNING
orWorker::IDLE
states.isRunning(): boolean
- true if worker is inWorker::RUNNING
state.isIdle(): boolean
- true if worker is inWorker::IDLE
state.getPid(): int
- returns process id of worker.getCurrentPayload(): int
- returns serial number of last done payload.
Warning about worker re-using! You can't restart a worker that has been terminated (with stop()
or kill()
), you need to create new worker and start it with start()
.
WorkersPool API
countIdleWorkers(): integer
- returns number of workers that are inWorker::IDLE
state.countRunningWorkers(): integer
- returns number of workers that are inWorker::RUNNING
state.countActiveWorkers(): integer
- returns number of workers that are either inWorker::RUNNING
orWorker::IDLE
states.getRunningWorkers(): Worker[]
- returns workers that are inWorker::RUNNING
state.enableDataOverhead()
/disableDataOverhead()
- enables/disables dataOverhead-mode.-
sendData($data, $wait = false): null|boolean
- dispatches payload to any free worker. Behavior depends on dataOverhead feature status:- When dataOverhead is disabled and
$wait = false
(by default), this method returnsnull
when no free workers available orboolean
with status of dispatching (true/false
). - When dataOverhead is disabled (by default) and
$wait = true
, this method will hold the execution of the script until any worker became free, dispatch your payload to it and return the status of dispatching (true/false
). - When dataOverhead is enabled, this method will dispatch your payload to any free worker. If there's not free workers, it will put new tasks in workers internal queues, which will be processed. This method uses fair distribution between all workers (so you can be sure that 24 tasks will be distributed between 6 workers as 4 per worker).
- When dataOverhead is disabled and
waitToFinish(array $trackers = null)
- holds the execution of script untill all workers go intoIDLE
state.
Predefined workers
DownloadWorker
As you've seen in examples, we created a downloading worker. But there is no need for this, we could use predefined DownloadWorker
which does the same.
- Full path:
wapmorgan\Threadable\DownloadWorker
- Description: downloads remote file and saves it on local server.
- Payload (array):
source
- remote file urltarget
- local file path
ExtractWorker
Zip-archives extractor.
- Full path:
wapmorgan\Threadable\ExtractWorker
- Description: extracts given zip archive to a folder
- Payload (array):
archive
- archive filenameoutput
- output directory
Use cases
Examples of programs that can be built with Threadable
:
- Media converters / encoders
- Data importers / exporters
- Bots for social networks / messengers
- Parsers / Scanners / Analyzers
- Servers (don't recommend unless you want to reinvent the wheel)
- ...