DataObjects/core.sql
[Pman.Core] / Notify.php
index 897ecfc..f7cbe57 100644 (file)
@@ -63,9 +63,40 @@ class Pman_Core_Notify extends Pman
             'min' => 0,
             'max' => 1,
         ),
+         'limit' => array(
+            'desc' => 'Limit search for no. to send to ',
+            'default' => 1000,
+            'short' => 'L',
+            'min' => 0,
+            'max' => 999,
+        ),
+        'dryrun' => array(
+            'desc' => 'Dry run - do not send.',
+            'default' => 0,
+            'short' => 'D',
+            'min' => 0,
+            'max' => 0,
+        ),
+        'poolsize' => array(
+            'desc' => 'Pool size',
+            'default' => 10,
+            'short' => 'P',
+            'min' => 0,
+            'max' => 100,
+        ),
     );
-    
-    
+    /**
+     * @var $nice_level Unix 'nice' level to stop it jamming server up.
+     */
+    var $nice_level = false;
+    /**
+     * @var $max_pool_size maximum runners to start at once.
+     */
+    var $max_pool_size = 10;
+    /**
+     * @var $max_to_domain maximum connections to make to a single domain
+     */
+    var $max_to_domain = 10;
     
     var $table = 'core_notify';
     var $target = 'Core/NotifySend';
@@ -91,6 +122,14 @@ class Pman_Core_Notify extends Pman
             DB_DataObject::debugLevel($opts['debug']);
             print_r($opts);
         }
+        $this->opts = $opts;
+        if (!empty($opts['poolsize'])) {
+            $this->max_pool_size = $opts['poolsize'];
+        }
+        
+        if (empty($opts['limit'])) {
+            $opts['limit'] = '1000'; // not sure why it's not picking up the defautl..
+        }
         //date_default_timezone_set('UTC');
        // phpinfo();exit;
         $showold = !empty($opts['old']);
@@ -127,7 +166,7 @@ class Pman_Core_Notify extends Pman
             // standard
             
             //$w->whereAdd('act_when > sent'); // eg.. sent is not valid..
-            $w->whereAdd("sent < '1970-01-01'"); // eg.. sent is not valid..
+            $w->whereAdd("sent < '1970-01-01' OR sent IS NULL"); // eg.. sent is not valid..
             
             $w->whereAdd('act_start > NOW() - INTERVAL 14 DAY'); // ignore the ones stuck in the queue
             if (!$this->force) {
@@ -135,10 +174,13 @@ class Pman_Core_Notify extends Pman
             }
     
             $w->orderBy('act_when ASC'); // oldest first.
-            $w->limit(1000); // we can run 1000 ...
+            
+            echo "QUEUE is {$w->count()}\n";
+            
+            $w->limit($opts['limit']); // we can run 1000 ...
         } else {
             $w->orderBy('act_when DESC'); // latest first
-            $w->limit(50); // we can run 1000 ...
+            $w->limit($opts['limit']); // we can run 1000 ...
         }
         if (!empty($this->evtype)) {
             $w->evtype = $this->evtype;
@@ -163,55 +205,81 @@ class Pman_Core_Notify extends Pman
             exit;
         }
         
-        
+        //echo "BATCH SIZE: ".  count($ar) . "\n";
         $pushed = array();
+        $requeue = array();
         while (true) {
+            
+            
+            echo "BATCH SIZE: ".  count($ar) . "\n";
+            
             if (empty($ar)) {
-                break;
+                echo "COMPLETED MAIN QUEUE - running delated\n";
+                
+                if (empty($pushed)) {
+                    break;
+                }
+                $ar = $pushed;
+                $pushed = false;
+                continue;
             }
             
             
             $p = array_shift($ar);
             if (!$this->poolfree()) {
                 array_unshift($ar,$p); /// put it back on..
-                //sleep(3);
+                sleep(3);
                 continue;
             }
-            if ($this->poolHasDomain($p->person_id_email)) {
-                if (in_array($p->person_id_email, $pushed)) {
-                    // it's been pushed to the end, and nothing has
-                    // been pushed since.
-                    // give up, let the next run sort it out.
-                    break;
-                }
+            if ($this->poolHasDomain($p->person_id_email) > $this->max_to_domain) {
                 
-                $ar[] = $p; // push it on the end..
+                if ($pushed === false) {
+                    // we only try once to requeue..
+                    $requeue[] = $p;
+                    continue;
+                }
+                $pushed[] = $p;
                 
-                $pushed[] = $p->person_id_email;
                 
-                echo "domain {$p->person_id_email} already on queue, pushing to end.\n";
                 //sleep(3);
                 continue;
             }
             
             
             $this->run($p->id,$p->person_id_email);
-            $pushed = array();
+            
+            
             
         }
         
         // we should have a time limit here...
         while(count($this->pool)) {
             $this->poolfree();
-            //sleep(3);
+             sleep(3);
+        }
+         
+        foreach($requeue as $p) {
+            $pp = clone($p);
+            $p->act_when = $p->sqlValue('NOW + INTERVAL 1 MINUTE');
+            $p->update($pp);
+            
         }
         
+        
+        
         die("DONE\n");
     }
     
     function run($id, $email, $cmdOpts="")
     {
-       // phpinfo();exit;
+        
+        static $renice = false;
+        if (!$renice) {
+            require_once 'System.php';
+            $renice = System::which('renice');
+        }
+        
+        // phpinfo();exit;
         $tn = tempnam(ini_get('session.save_path'),'stdout') . '.stdout';
         $descriptorspec = array(
             0 => array("pipe", 'r'),  // stdin is a pipe that the child will read from
@@ -234,8 +302,26 @@ class Pman_Core_Notify extends Pman
        
         $pipe = array();
         echo "call proc_open $cmd\n";
+        
+        
+        if ($this->max_pool_size === 1) {
+            passthru($cmd);
+            return;
+        }
+        
+        
+        if (!empty($this->opts['dryrun'])) {
+            echo "DRY RUN\n";
+            return;
+        }
+        
         $p = proc_open($cmd, $descriptorspec, $pipes, $cwd );
         $info =  proc_get_status($p);
+        
+        if ($this->nice_level !== false) { 
+            $rcmd = "$renice {$this->nice_level} {$info['pid']}";
+            `$rcmd`;
+        } 
         $this->pool[] = array(
                 'proc' => $p,
                 'pid' => $info['pid'],
@@ -263,12 +349,12 @@ class Pman_Core_Notify extends Pman
             //var_dump($info);
             
             // update if necessday.
-            if ($info['pid']) {
+            if ($info['pid'] && $p['pid'] != $info['pid']) {
                 echo "CHANING PID FROM " . $p['pid']  .  "  TO ". $info['pid']. "\n";
                 $p['pid'] = $info['pid'];
             }
             
-            echo @file_get_contents('/proc/'. $p['pid'] .'/cmdline') . "\n";
+            //echo @file_get_contents('/proc/'. $p['pid'] .'/cmdline') . "\n";
             
             if ($info['running']) {
             
@@ -309,7 +395,7 @@ class Pman_Core_Notify extends Pman
         }
         echo "POOL SIZE: ". count($pool) ."\n";
         $this->pool = $pool;
-        if (count($pool) < 10) {
+        if (count($pool) < $this->max_pool_size) {
             return true;
         }
         return false;
@@ -322,14 +408,15 @@ class Pman_Core_Notify extends Pman
      */
     function poolHasDomain($email)
     {
+        $ret = 0;
         $dom = strtolower(array_pop(explode('@',$email)));
         foreach($this->pool as $p) {
             $mdom = strtolower(array_pop(explode('@',$p['email'])));
             if ($mdom == $dom) {
-                return true;
+                $ret++;
             }
         }
-        return false;
+        return $ret;
         
     }