/**
  * worker master - Pool
  */
 function workerPool()
 {
     $start = microtime(true);
     echo "\nworker start at {$start}";
     $jobs = array();
     if ($jobs) {
         $wNum = 5;
         //进程数
         $jobCnt = count($jobs);
         $w = ceil($jobCnt / $wNum);
         //
         $pool = new Pool("demoWorker", $w);
         $pool->addJobs($jobs);
         $pool->waiting();
         $stop = microtime(true);
         $cost = $stop - $start;
         echo "\nworker {$jobCnt} jobs done at {$stop}";
         echo "\nworker cost time = {$cost}";
     } else {
         //for daemon mode
         echo "\nworker no job and sleep 5s";
         sleep(5);
     }
     $end = microtime(true);
     echo "\nworker end at {$end}";
 }

 protected function assertHit($expected, Pool $pool, $key)
 {
     $this->assertEquals($expected, $pool->get($key));
     $this->assertEquals($expected, $pool->getItem($key)->get());
     $this->assertTrue($pool->getItem($key)->isHit());
     $this->assertTrue($pool->hasItem($key));
 }

 /**
  * Called when the worker is ready to go
  * @return void
  */
 public function onReady()
 {
     $this->redis = \PHPDaemon\Clients\Redis\Pool::getInstance();
     $this->redis->multi(function ($multi) {
         // "OK"
         D('start multi: ' . $multi->result);
         $multi->set('test1', 'value1', function ($redis) use($multi) {
             // "QUEUED"
             D('in multi 1: ' . $redis->result);
             $this->redis->set('test1', 'value1-new', function ($redis) {
                 // "OK", not "QUEUED"
                 D('out multi 1: ' . $redis->result);
             });
             setTimeout(function ($timer) use($multi) {
                 // "QUEUED"
                 $multi->set('test2', 'value2', function ($redis) use($multi) {
                     D('in multi 2: ' . $redis->result);
                     $multi->exec(function ($redis) {
                         D('exec');
                         D($redis->result);
                     });
                 });
                 $timer->free();
             }, 200000.0);
         });
     });
 }

 public function testPoolGc()
 {
     $pool = new Pool(1, PoolTestWorker::class, [new stdClass(), new Threaded()]);
     $work = new PoolTestWork();
     $pool->submit($work);
     while (@$i++ < 2) {
         $pool->submit(new PoolTestWork());
         # nothing to assert, no exceptions please
     }
     $pool->submitTo(0, new PoolTestWork());
     # nothing to assert, no exceptions please
     /* synchronize with pool */
     $sync = new PoolTestSync();
     $pool->submit($sync);
     $sync->synchronized(function ($sync) {
         if (!$sync->finished) {
             $sync->wait();
         }
     }, $sync);
     $pool->collect(function ($task) {
         $this->assertTrue($task->isGarbage());
         return true;
     });
     $pool->shutdown();
 }

 public function useMultiCore1()
 {
     $core = $this->getCoreCount();
     $pool = new \Pool($core);
     for ($i = 0; $i < $core; $i++) {
         $pool->submit(new ProveWorker());
     }
 }

 /**
  * @param string[] $adjustments
  * @return AdjustmentInterface[]
  */
 protected function fetchAdjustments($adjustments)
 {
     $instances = [];
     foreach ($adjustments as $code) {
         $instances[$code] = $this->adjustmentPool->getAdjustmentByCode($code);
     }
     uasort($instances, [$this, 'sortAdjustments']);
     return $instances;
 }

 function allAction()
 {
     $order = addslashes(trim($this->_request->getParam('order', 'Name')));
     $this->view->title = $this->view->translate->_("Pools");
     // to view
     $this->view->meta_refresh = 300;
     // meta http-equiv="refresh"
     $pools = new Pool();
     $this->view->pools = $pools->fetchAll(null, $order);
 }

 protected function execute(InputInterface $input, OutputInterface $output)
 {
     $pool = new \Pool($input->getOption('threads'), \Worker::class);
     foreach ($input->getArgument('indexes') as $index) {
         $pool->submit(new IndexerRunner($input->getOption('indexer'), $input->getOption('config'), $index));
     }
     $pool->shutdown();
     $pool->collect(function (IndexerRunner $work) use($output) {
         $output->writeln($work->return);
     });
 }

 public function __construct($threadSayisi, BurtayThread $sinif)
 {
     $pool = new Pool($threadSayisi);
     $is_listesi = array();
     for ($i = 1; $i <= $threadSayisi; $i++) {
         $is_listesi[] = new burtayThreadPool($sinif);
     }
     foreach ($is_listesi as $liste) {
         $pool->submit($liste);
     }
     $pool->shutdown();
 }

 /**
  * @param Pool $pool
  *
  * @return array|bool
  */
 private function shuffleTillMatch(Pool $pool)
 {
     $timeToStop = microtime(true) + self::SHUFFLE_TIME_LIMIT;
     $entries = $pool->getEntries()->getValues();
     while (microtime(true) < $timeToStop) {
         $set = $this->shuffleArray($entries);
         if ($this->checkValidMatch($entries, $set)) {
             $this->matchedExcludes = $set;
             return $set;
         }
     }
     return false;
 }

 public function testUnsubscribeAll()
 {
     $pool = new Pool();
     $mockEvent1 = $this->getMockBuilder('Athrois\\Event')->getMock();
     $mockEvent2 = $this->getMockBuilder('AnotherEvent')->getMock();
     $mockListener = $this->getMockBuilder('Athrois\\Listener')->getMock();
     $identifier1 = Pool::getId($mockEvent1);
     $identifier2 = Pool::getId($mockEvent2);
     $pool->register($mockListener, $identifier1);
     $pool->register($mockListener, $identifier2);
     $pool->unsubscribe($mockListener);
     $this->assertAttributeEquals(array(), 'listeners', $pool);
 }

 /**
  * Called when the worker is ready to go.
  * @return void
  */
 public function onReady()
 {
     $this->redis = \PHPDaemon\Clients\Redis\Pool::getInstance();
     /*$this->redis->eval("return {'a','b','c', {'d','e','f', {'g','h','i'}} }",0, function($redis) {
     			Daemon::log(Debug::dump($redis->result));
     		});*/
     $this->redis->subscribe('te3st', function ($redis) {
         Daemon::log(Debug::dump($redis->result));
     });
     $this->redis->psubscribe('test*', function ($redis) {
         Daemon::log(Debug::dump($redis->result));
     });
 }

 public function setPool()
 {
     $pool = new Pool();
     foreach ($this->getConfiguration() as $key => $value) {
         $array = new ArrayObject($value);
         $server = new Server();
         $server->setHost($array->offsetGet(self::SERVER_PROPERTY_HOST))->setPort($array->offsetGet(self::SERVER_PROPERTY_PORT))->setName($key);
         if ($array->offsetExists(self::SERVER_PROPERTY_AUTH)) {
             $server->setAuth($array->offsetGet(self::SERVER_PROPERTY_AUTH));
         }
         $pool->attach($server);
     }
     $this->pool = $pool;
 }

 /**
  * @param int $currentTick
  */
 public function mainThreadHeartbeat($currentTick)
 {
     $this->currentTick = $currentTick;
     while ($this->isReady($this->currentTick)) {
         /** @var TaskHandler $task */
         $task = $this->queue->extract();
         if ($task->isCancelled()) {
             unset($this->tasks[$task->getTaskId()]);
             continue;
         } else {
             $task->timings->startTiming();
             $task->run($this->currentTick);
             $task->timings->stopTiming();
         }
         if ($task->isRepeating()) {
             $task->setNextRun($this->currentTick + $task->getPeriod());
             $this->queue->insert($task, $this->currentTick + $task->getPeriod());
         } else {
             $task->remove();
             unset($this->tasks[$task->getTaskId()]);
         }
     }
     if ($this->asyncTasks > 0) {
         //Garbage collector
         $this->asyncPool->collect([$this, "collectAsyncTask"]);
         foreach ($this->asyncTaskStorage as $asyncTask) {
             if ($asyncTask->isFinished() and !$asyncTask->isCompleted()) {
                 $this->collectAsyncTask($asyncTask);
             }
         }
     }
 }

 private function testPool()
 {
     $initTime = microtime(true);
     $pool = new \Pool(8, \Worker::class);
     for ($i = 0; $i < 10; $i++) {
         $pool->submit(new Work());
     }
     $initTime = microtime(true) - $initTime;
     $finishTime = microtime(true);
     $pool->shutdown();
     $pool->collect(function (Work $work) {
         echo 'Work of ', get_class($work), ' #', $work->i, ' - ', $work->getWork(), PHP_EOL;
     });
     $finishTime = microtime(true) - $finishTime;
     return [$initTime, $finishTime];
 }

 public function definePool($name, $num = null)
 {
     $parts = explode(':', $name);
     $alias = $parts[0];
     if (count($parts) > 1) {
         $class = $parts[1];
     } else {
         $class = $parts[0];
     }
     if (!$class) {
         $class = $alias;
     }
     $this->pools[$alias] = $pool = new Pool(new Definition($class), $num);
     $pool->setFactory($this);
     return $pool->getDefinition();
 }

 public function get_dataksmurni($date = false)
 {
     if (!$date) {
         $date = date('Y-m-d');
     }
     $month = date('n', strtotime($date));
     $year = date('Y', strtotime($date));
     $series = array();
     $categories = array();
     foreach (Pool::all() as $pool) {
         $datafianan = array();
         for ($j = 1; $j <= date('t', strtotime($date)); $j++) {
             $categories[] = $j;
             $datafianan[] = 0;
         }
         $reportdaily = DB::table('financial_report_graf_ks_detail')->where_month($month)->where_year($year)->order_by('operasi_time', 'asc')->where_pool_id($pool->id)->where_in('operasi_status_id', array(1, 3))->where('setoran_cash', '<', 50000000)->group_by('operasi_time')->get(array('operasi_time', DB::raw('sum(ks) as subks')));
         $arraydata = array();
         foreach ($reportdaily as $finan) {
             $potition = date('j', strtotime($finan->operasi_time)) - 1;
             $arraydata[$potition] = (int) $finan->subks;
         }
         $datas = array_replace($datafianan, $arraydata);
         $series[] = array('name' => $pool->pool_name, 'data' => $datas);
     }
     $returndata = array('categories' => $categories, 'series' => $series);
     return json_encode($returndata);
 }

 public function get_add()
 {
     $this->data['pools'] = Koki::to_dropdown(Pool::all(), 'id', 'pool_name', array(0 => 'All'));
     $this->data['users'] = Koki::to_dropdown(User::all(), 'id', 'first_name');
     $this->data['create'] = true;
     return View::make('themes.modul.' . $this->views . '.form', $this->data);
 }

 public static function wrap($coroutine)
 {
     if ($coroutine instanceof \Generator) {
         $coKey = $coroutine->key();
         $coValue = $coroutine->current();
         if ($coKey && in_array($coKey, self::$_ioQueue)) {
             try {
                 $client = \Pool::get($coKey, $coValue);
                 if ($client) {
                     $client->setCoroutine($coroutine);
                     $coroutine->send($client);
                 } else {
                     Queue::push($coKey, $coroutine);
                 }
             } catch (exception $e) {
                 $coroutine->throw($e);
             }
         } else {
             if ($coValue instanceof \Coroutine\Base) {
                 $coValue->setCoroutine($coroutine);
             }
         }
     }
     return $coroutine;
 }

 public function get_datamonthly($date = false)
 {
     if (!$date) {
         $date = date('Y-m-d');
     }
     $year = date('Y', strtotime($date));
     $series = array();
     $categories = array();
     foreach (Pool::all() as $pool) {
         $datafianan = array();
         for ($j = 1; $j <= 12; $j++) {
             $categories[] = $j;
             $datafianan[] = 0;
         }
         $reportmonthly = DB::table('financial_report_summary_month_graf')->where_year($year)->order_by('operasi_time', 'asc')->where_pool_id($pool->id)->where('setoran_cash', '<', 993000000)->get();
         $arraydata = array();
         $bulan = 0;
         foreach ($reportmonthly as $finan) {
             $potition = $finan->month;
             $arraydata[$potition] = (int) ($finan->setoran_cash - ($finan->biaya_cuci + $finan->iuran_laka));
         }
         $datas = array_replace($datafianan, $arraydata);
         $series[] = array('name' => $pool->pool_name, 'data' => $datas);
     }
     $returndata = array('categories' => $categories, 'series' => $series);
     return json_encode($returndata);
 }