Issue #134 Run as daemon with workers

scheduler_cron.sh

@@ -40,6 +40,9 @@ INCLUDE_GROUPS=""
 EXCLUDE_GROUPS=""
 INCLUDE_JOBS=""
 EXCLUDE_JOBS=""
+ACTION=""
+WORKERS=10
+WAIT=1
 
 # Parse command line args (very simplistic)
 while [ $# -gt 0 ]; do
@@ -64,6 +67,18 @@ while [ $# -gt 0 ]; do
             EXCLUDE_JOBS=$2
             shift 2
         ;;
+        --action)
+            ACTION=$2
+            shift 2
+        ;;
+        --workers)
+            WORKERS=$2
+            shift 2
+        ;;
+        --wait)
+            WAIT=$2
+            shift 2
+        ;;
         --)
             shift
             break
@@ -81,6 +96,75 @@ if [ -z "${MODE}" ]; then
     exit 1
 fi
 
+if [ "${MODE}" = "daemonize" ]; then
+    cd "${DIR}"
+    while true; do
+        # Count the number of running jobs right now
+        RUNNING_WORKERS=`ps aux | grep "${SCHEDULER} --action daemonRun" | grep -v grep | wc -l`
+        # Only fetch the next job if there are enough available workers
+        if [ "${RUNNING_WORKERS}" -lt ${WORKERS} ]; then
+            # Fetches the next schedule --code and --id to run
+            NEXT_JOB=`${PHP_BIN} ${SCHEDULER} --action daemonNext`
+            if [ ! -z "${NEXT_JOB}" ]; then
+                # Strip the --id param so we can search for jobs with the same code
+                NEXT_WITHOUT_ID=`echo $NEXT_JOB | sed -r 's/--id [0-9]+//g'`
+                # Only run this schedule now if a worker is not running a job with the same code
+                if ! ps auxwww | grep "${SCHEDULER} --action daemonRun ${NEXT_WITHOUT_ID}" | grep -v grep 1>/dev/null 2>/dev/null ; then
+                    "${PHP_BIN}" "${SCHEDULER}" --action runScheduleById ${NEXT_JOB} &
+                else
+                    # Otherwise mark it as missed
+                    "${PHP_BIN}" "${SCHEDULER}" --action daemonMultiple ${NEXT_JOB} &
+                fi
+            fi
+        else
+            sleep "${WAIT}"
+        fi
+    done
+fi
+
+function getDaemonPid {
+    DAEMON_PID=`ps aux | grep "$0 --mode daemonize" | grep -v grep | awk '{print $2}'`
+}
+
+if [ "${MODE}" = "daemon" ]; then
+    getDaemonPid
+    case "${ACTION}" in
+        start)
+            if [ ! -z "${DAEMON_PID}" ]; then
+                echo "Daemon is already running"
+            else
+                $0 --mode daemonize &
+                getDaemonPid
+                echo "Daemon started, PID ${DAEMON_PID}"
+            fi
+        ;;
+        stop)
+            if [ -z "${DAEMON_PID}" ]; then
+                echo "Daemon is not running"
+            else
+                kill "${DAEMON_PID}"
+                getDaemonPid
+                if [ -z "${DAEMON_PID}" ]; then
+                    echo "Daemon stopped"
+                else
+                    echo "Failed to stop daemon"
+                fi
+            fi
+        ;;
+        status)
+            if [ ! -z "${DAEMON_PID}" ]; then
+                echo "Daemon is running, PID ${DAEMON_PID}"
+            else
+                echo "Daemon is not running"
+            fi
+        ;;
+        *)
+            echo "Usage: ./scheduler_cron.sh --mode daemon --action (start|stop|status)"
+        ;;
+    esac
+    exit 0
+fi
+
 # Unique identifier for this cron job run
 IDENTIFIER=$(echo -n "${DIR}|${MODE}|${INCLUDE_GROUPS}|${EXCLUDE_GROUPS}|${INCLUDE_JOBS}|${EXCLUDE_JOBS}" | "${MD5SUM_BIN}" - | cut -f1 -d' ')
 

shell/scheduler.php

@@ -383,6 +383,69 @@ public function cronAction()
         }
     }
 
+    /**
+     * Get the next job to be processed by the daemon
+     *
+     * @return void
+     */
+    public function daemonNextAction()
+    {
+        Mage::getSingleton('aoe_scheduler/schedulemanager')->generateSchedules();
+        $next = Mage::getModel('cron/schedule')->getCollection()
+            ->addFieldToFilter('scheduled_at', array('lteq' => now()))
+            ->addFieldToFilter('status', Mage_Cron_Model_Schedule::STATUS_PENDING)
+            ->setOrder('scheduled_at', Zend_Db_Select::SQL_ASC)
+            ->getFirstItem();
+        if (!$next || !$next->getId()) {
+            return;
+        }
+        $dateNow = date_create(now());
+        $dateSchedule = date_create($next->getScheduledAt());
+        $diff = date_diff($dateNow, $dateSchedule);
+        if ($diff->format('%i') > Mage::getStoreConfig('system/cron/schedule_lifetime')) {
+            try {
+                $next->load()->setStatus(Mage_Cron_Model_Schedule::STATUS_MISSED)
+                    ->save();
+            } catch (Exception $e) {
+                Mage::logException($e->getMessage());
+            }
+            return;
+        }
+        echo '--code ' . $next->getJobCode() . ' ' . '--id ' . $next->getId();
+    }
+
+    /**
+     * Run a job specified by the ID
+     *
+     * @return void
+     */
+    public function runScheduleByIdAction()
+    {
+        $id = $this->getArg('id');
+        if (!$id) {
+            return;
+        }
+        Mage::getModel('cron/schedule')->load($id)->runNow();
+    }
+
+    /**
+     * Mark when multiple jobs with the same code try to run
+     *
+     * @return void
+     */
+    public function daemonMultipleAction()
+    {
+        $id = $this->getArg('id');
+        if (!$id) {
+            return;
+        }
+        Mage::getModel('cron/schedule')
+            ->load($id)
+            ->setMessages('Multiple tasks with the same job code were piling up. Skipping execution of duplicates.')
+            ->setStatus(Mage_Cron_Model_Schedule::STATUS_MISSED)
+            ->save();
+    }
+
     /**
      * Display extra help
      *