<?php
    class Threader
    {
        private $pid;
        private $socket;
        private $up_to;
        private static $inst = NULL;
        const MESSAGE_LENGTH = 10;

        private function Threader()
        {
            $this->pid = FALSE;
            $this->socket  = NULL;
            $this->up_to  = 0;
        }

        public static function current()
        {
            if(self::$inst == NULL)
                self::$inst = new Threader();

            return self::$inst;
        }

        public function countToOneMillion()
        {
            $sockets = array();

            if (socket_create_pair(AF_UNIX, SOCK_STREAM, 0,$sockets) === false)
                echo "socket_create_pair failed. Reason: ".socket_strerror(socket_last_error());

            $this->socket = $sockets[1];
            $this->pid = pcntl_fork();

            if($this->pid == -1)
                throw new Exception('Could not fork');

            else if(!$this->pid)
            {
                //THIS IS THE CHILD PROCESS
                if(socket_write($sockets[0],str_pad("RUNNING",
                                Threader::MESSAGE_LENGTH),
                                Threader::MESSAGE_LENGTH)
                                === false)
                    throw new Exception("socket_write() failed. Reason: ".
                              socket_strerror(socket_last_error($sockets[0])));

                for($i = 0;$i < 100000;$i++)
                {
                    if (socket_write($sockets[0], str_pad("UPTO $i\n", Threader::MESSAGE_LENGTH),
                        Threader::MESSAGE_LENGTH) === false)
                        throw new Exception("socket_write() failed. Reason: ".
                                  socket_strerror (socket_last_error($sockets[0])));
                }

                if(socket_write($sockets[0],str_pad("STOP", Threader::MESSAGE_LENGTH),
                   Threader::MESSAGE_LENGTH) === false)
                    throw new Exception("socket_write() failed.  Reason: ".
                              socket_strerror(socket_last_error($sockets[0])));

                socket_close($sockets[0]);
            }
        }

        public function pid()
        {
            return $this->pid;
        }

        public function running()
        {
            if($this->pid)
            {
                $ret = FALSE;
                //THIS IS THE PARENT PROCESS
                if(($data = socket_read($this->socket, Threader::MESSAGE_LENGTH, PHP_BINARY_READ))
                   !== false)
                {
                    $data = trim($data);

                    if(strpos($data,"RUNNING")!==FALSE)
                        $ret = true;

                    else if(strpos($data,"STOP")!==FALSE)
                    {
                        $ret = false;
                        socket_close($this->socket);
                    }

                    else
                    {
                        $this->up_to = str_replace('UPTO ','',
                                                   $data);
                        $ret = true;
                    }
                }

                return $ret;
            }
        }

        public function upTo()
        {
            if($this->pid)
                return $this->up_to;
        }
    }

    Threader::current()->countToOneMillion();

    if(Threader::current()->pid())
    {
        $old = 0;

        while($data = Threader::current()->running())
        {
            $up_to = Threader::current()->upTo();

            if(($up_to - $old)>1000)
            {
                echo($up_to."\n");
                $old = $up_to;
            }
        }

        echo("All done!\n");
    }