<commit_before><?php

/**
 * This class centralizes job suspension utilities
 * 
 * @package Core
 * @subpackage Batch
 *
 */
class kJobsSuspender {
	
	/** The maximum number of jobs we will handle for each (dc,partner_id,job_type,job_sub_type) */
	const MAX_PROCESSED_ROWS = 500;
	
	/**
	 * Entry point to job balancing.
	 * - Move jobs from status '0' to status '11' in case there are too many pending jobs
	 * - move jobs back from status '11' to status '0' in case the load is over.
	 * 
	 */
	public static function balanceJobsload() {
	
		$minPendingJobs = self::getSuspenderMinPendingJobs();
		$maxPendingJobs = self::getSuspenderMaxPendingJobs();
		
		$dcId = kDataCenterMgr::getCurrentDcId();
		$loadedKeys = array();
	
		// Suspend Jobs
		$c = self::createJobBalanceQuery($dcId, $minPendingJobs);
		$stmt = BatchJobLockPeer::doSelectStmt($c);
		$rows= $stmt->fetchAll(PDO::FETCH_ASSOC);
	
		foreach ($rows as $row) {
			$partnerId = $row['PARTNER_ID'];
			$jobType = $row['JOB_TYPE'];
			$jobSubType = $row['JOB_SUB_TYPE'];
				
			$loadedKeys[] = $partnerId . "#" . $jobType . "#" . $jobSubType;
			$jobCount = $row[BatchJobLockPeer::COUNT];

			if($jobCount > $maxPendingJobs) {
				self::suspendJobs(($jobCount - $maxPendingJobs), $dcId, $partnerId, $jobType, $jobSubType);
			}
		}
	
		// Unsuspend jobs
		$c = self::createReturnBalancedJobsQuery($dcId);
		$stmt = BatchJobLockSuspendPeer::doSelectStmt($c);
		$rows= $stmt->fetchAll(PDO::FETCH_ASSOC);
	
		foreach ($rows as $row) {
			$partnerId = $row['PARTNER_ID'];
			$jobType = $row['JOB_TYPE'];
			$jobSubType = $row['JOB_SUB_TYPE'];
			$key = $partnerId . "#" . $jobType . "#" . $jobSubType;
				
			if(!in_array($key, $loadedKeys))
				self::unsuspendJobs(($maxPendingJobs - $minPendingJobs), $dcId, $partnerId, $jobType, $jobSubType);
		}
	}
	
	/**
	 * This function generates the query that gets all the jobs that needs to be suspended.
	 */
	private static function createJobBalanceQuery($dc, $value) {
		$c = new Criteria();
	
		// Where
		$c->add(BatchJobLockPeer::STATUS, BatchJob::BATCHJOB_STATUS_PENDING, Criteria::EQUAL);
		$c->add(BatchJobLockPeer::DC, $dc, Criteria::EQUAL);
		$c->add(BatchJobLockPeer::SCHEDULER_ID, null, Criteria::ISNULL);
		// Group by
		$c->addGroupByColumn(BatchJobLockPeer::DC);
		$c->addGroupByColumn(BatchJobLockPeer::PARTNER_ID);
		$c->addGroupByColumn(BatchJobLockPeer::JOB_TYPE);
		$c->addGroupByColumn(BatchJobLockPeer::JOB_SUB_TYPE);
		// Having
		$c->addHaving($c->getNewCriterion(BatchJobLockPeer::ID, BatchJobLockPeer::COUNT . '>' . $value, Criteria::CUSTOM));
		// Select
		$c->addSelectColumn(BatchJobLockPeer::COUNT);
		foreach($c->getGroupByColumns() as $column)
			$c->addSelectColumn($column);
		return $c;
	}
	
	/**
	 * This function generates the query that gets all the jobs that needs to be unsuspended.
	 * All those jobs will appear in batch_job_lock_suspend table
	 */
	private static function createReturnBalancedJobsQuery($dc) {
		$c = new Criteria();
	
		// Where
		$c->add(BatchJobLockSuspendPeer::DC, $dc, Criteria::EQUAL);
		// Group by
		$c->addGroupByColumn(BatchJobLockSuspendPeer::DC);
		$c->addGroupByColumn(BatchJobLockSuspendPeer::PARTNER_ID);
		$c->addGroupByColumn(BatchJobLockSuspendPeer::JOB_TYPE);
		$c->addGroupByColumn(BatchJobLockSuspendPeer::JOB_SUB_TYPE);
		foreach($c->getGroupByColumns() as $column)
			$c->addSelectColumn($column);
		return $c;
	}
	
	/**
	 * This function suspends up to ($limit) jobs of a given ($dc, $partnerId, $jobType, $jobSubType) 
	 */
	private static function suspendJobs($limit, $dc, $partnerId, $jobType, $jobSubType) {
	
		$limit = min(self::MAX_PROCESSED_ROWS, $limit);
		
		// Find IDs
		$c = new Criteria();
		$c->addSelectColumn(BatchJobLockPeer::ID);
		$c->add( BatchJobLockPeer::STATUS, BatchJob::BATCHJOB_STATUS_PENDING, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::DC, $dc, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::PARTNER_ID, $partnerId, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::JOB_TYPE, $jobType, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::JOB_SUB_TYPE, $jobSubType, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::SCHEDULER_ID, null, Criteria::ISNULL);
	
		$c->addDescendingOrderByColumn(BatchJobLockPeer::PRIORITY);
		$c->addDescendingOrderByColumn(BatchJobLockPeer::URGENCY);
		$c->addDescendingOrderByColumn(BatchJobLockPeer::ESTIMATED_EFFORT);
		$c->setLimit($limit);
	
		$stmt = BatchJobLockPeer::doSelectStmt($c);
		$jobIds = $stmt->fetchAll(PDO::FETCH_COLUMN);
	
		// Suspend chosen ids
		$res = self::setJobsStatus($jobIds, BatchJob::BATCHJOB_STATUS_SUSPEND, BatchJob::BATCHJOB_STATUS_PENDING, true);
		$rootJobIds = self::suspendRootJobs($jobIds);
		self::moveToSuspendedJobsTable(array_merge($jobIds, $rootJobIds));
		KalturaLog::info("$res jobs of partner ($partnerId) job type ($jobType / $jobSubType) on DC ($dc) were suspended");
		KalturaLog::info("As a result, ". count($rootJobIds) . " root jobs were suspended.");
	}
	
	/**
	 * This function unsuspends up to ($limit) jobs of a given ($dc, $partnerId, $jobType, $jobSubType)
	 */
	private static function unsuspendJobs($limit, $dc, $partnerId, $jobType, $jobSubType) {
	
		$limit = min(self::MAX_PROCESSED_ROWS, $limit);
		
		// Find IDs from Batch Job Suspend Table
		$c = new Criteria();
		$c->addSelectColumn(BatchJobLockSuspendPeer::ID);
		$c->add( BatchJobLockSuspendPeer::DC, $dc, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::PARTNER_ID, $partnerId, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::JOB_TYPE, $jobType, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::JOB_SUB_TYPE, $jobSubType, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::STATUS, BatchJob::BATCHJOB_STATUS_SUSPEND);
	
		$c->addAscendingOrderByColumn(BatchJobLockSuspendPeer::PRIORITY);
		$c->addAscendingOrderByColumn(BatchJobLockSuspendPeer::URGENCY);
		$c->addAscendingOrderByColumn(BatchJobLockSuspendPeer::ESTIMATED_EFFORT);
		$c->setLimit($limit);
	
		$stmt = BatchJobLockSuspendPeer::doSelectStmt($c);
		$jobIds = $stmt->fetchAll(PDO::FETCH_COLUMN);
	
		// Return the jobs from batch_job_lock_suspend table
		self::moveFromSuspendedJobsTable($jobIds);
		$rootJobIds = self::unsuspendRootJob($jobIds);
		// Update the jobs status to pending
		$res = self::setJobsStatus($jobIds, BatchJob::BATCHJOB_STATUS_PENDING, BatchJob::BATCHJOB_STATUS_SUSPEND, false);
		$resRoot = self::setJobsStatus($rootJobIds, BatchJob::BATCHJOB_STATUS_ALMOST_DONE, BatchJob::BATCHJOB_STATUS_SUSPEND_ALMOST_DONE, false);
		KalturaLog::info("$res jobs of partner ($partnerId) job type ($jobType / $jobSubType) on DC ($dc) were unsuspended");
		KalturaLog::info("As a result $resRoot were unsuspended.");
	}
	
	/**
	 * This function sets the status of the given $jobIds to the given $status.
	 * We're saved from the kFlowManager::updatedJob callback since we work directly through the DB. 
	 * No callbacks are called.
	 * Returns the number of the affected rows.
	 */
	private static function setJobsStatus($jobIds, $status, $oldStatus, $addSchedulerCond) {
	
		$suspenderUpdateChunk = self::getSuspenderUpdateChunk();
		
		$affectedRows = 0;<commit_msg>SUP-25084: unsuspend jobs with status suspend_almost_done<commit_after><?php

/**
 * This class centralizes job suspension utilities
 * 
 * @package Core
 * @subpackage Batch
 *
 */
class kJobsSuspender {
	
	/** The maximum number of jobs we will handle for each (dc,partner_id,job_type,job_sub_type) */
	const MAX_PROCESSED_ROWS = 500;
	
	/**
	 * Entry point to job balancing.
	 * - Move jobs from status '0' to status '11' in case there are too many pending jobs
	 * - move jobs back from status '11' to status '0' in case the load is over.
	 * 
	 */
	public static function balanceJobsload() {
	
		$minPendingJobs = self::getSuspenderMinPendingJobs();
		$maxPendingJobs = self::getSuspenderMaxPendingJobs();
		
		$dcId = kDataCenterMgr::getCurrentDcId();
		$loadedKeys = array();
	
		// Suspend Jobs
		$c = self::createJobBalanceQuery($dcId, $minPendingJobs);
		$stmt = BatchJobLockPeer::doSelectStmt($c);
		$rows= $stmt->fetchAll(PDO::FETCH_ASSOC);
	
		foreach ($rows as $row) {
			$partnerId = $row['PARTNER_ID'];
			$jobType = $row['JOB_TYPE'];
			$jobSubType = $row['JOB_SUB_TYPE'];
				
			$loadedKeys[] = $partnerId . "#" . $jobType . "#" . $jobSubType;
			$jobCount = $row[BatchJobLockPeer::COUNT];

			if($jobCount > $maxPendingJobs) {
				self::suspendJobs(($jobCount - $maxPendingJobs), $dcId, $partnerId, $jobType, $jobSubType);
			}
		}
	
		// Unsuspend jobs
		$c = self::createReturnBalancedJobsQuery($dcId);
		$stmt = BatchJobLockSuspendPeer::doSelectStmt($c);
		$rows= $stmt->fetchAll(PDO::FETCH_ASSOC);
	
		foreach ($rows as $row) {
			$partnerId = $row['PARTNER_ID'];
			$jobType = $row['JOB_TYPE'];
			$jobSubType = $row['JOB_SUB_TYPE'];
			$key = $partnerId . "#" . $jobType . "#" . $jobSubType;
				
			if(!in_array($key, $loadedKeys))
				self::unsuspendJobs(($maxPendingJobs - $minPendingJobs), $dcId, $partnerId, $jobType, $jobSubType);
		}
	}
	
	/**
	 * This function generates the query that gets all the jobs that needs to be suspended.
	 */
	private static function createJobBalanceQuery($dc, $value) {
		$c = new Criteria();
	
		// Where
		$c->add(BatchJobLockPeer::STATUS, BatchJob::BATCHJOB_STATUS_PENDING, Criteria::EQUAL);
		$c->add(BatchJobLockPeer::DC, $dc, Criteria::EQUAL);
		$c->add(BatchJobLockPeer::SCHEDULER_ID, null, Criteria::ISNULL);
		// Group by
		$c->addGroupByColumn(BatchJobLockPeer::DC);
		$c->addGroupByColumn(BatchJobLockPeer::PARTNER_ID);
		$c->addGroupByColumn(BatchJobLockPeer::JOB_TYPE);
		$c->addGroupByColumn(BatchJobLockPeer::JOB_SUB_TYPE);
		// Having
		$c->addHaving($c->getNewCriterion(BatchJobLockPeer::ID, BatchJobLockPeer::COUNT . '>' . $value, Criteria::CUSTOM));
		// Select
		$c->addSelectColumn(BatchJobLockPeer::COUNT);
		foreach($c->getGroupByColumns() as $column)
			$c->addSelectColumn($column);
		return $c;
	}
	
	/**
	 * This function generates the query that gets all the jobs that needs to be unsuspended.
	 * All those jobs will appear in batch_job_lock_suspend table
	 */
	private static function createReturnBalancedJobsQuery($dc) {
		$c = new Criteria();
	
		// Where
		$c->add(BatchJobLockSuspendPeer::DC, $dc, Criteria::EQUAL);
		// Group by
		$c->addGroupByColumn(BatchJobLockSuspendPeer::DC);
		$c->addGroupByColumn(BatchJobLockSuspendPeer::PARTNER_ID);
		$c->addGroupByColumn(BatchJobLockSuspendPeer::JOB_TYPE);
		$c->addGroupByColumn(BatchJobLockSuspendPeer::JOB_SUB_TYPE);
		foreach($c->getGroupByColumns() as $column)
			$c->addSelectColumn($column);
		return $c;
	}
	
	/**
	 * This function suspends up to ($limit) jobs of a given ($dc, $partnerId, $jobType, $jobSubType) 
	 */
	private static function suspendJobs($limit, $dc, $partnerId, $jobType, $jobSubType) {
	
		$limit = min(self::MAX_PROCESSED_ROWS, $limit);
		
		// Find IDs
		$c = new Criteria();
		$c->addSelectColumn(BatchJobLockPeer::ID);
		$c->add( BatchJobLockPeer::STATUS, BatchJob::BATCHJOB_STATUS_PENDING, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::DC, $dc, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::PARTNER_ID, $partnerId, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::JOB_TYPE, $jobType, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::JOB_SUB_TYPE, $jobSubType, Criteria::EQUAL);
		$c->add( BatchJobLockPeer::SCHEDULER_ID, null, Criteria::ISNULL);
	
		$c->addDescendingOrderByColumn(BatchJobLockPeer::PRIORITY);
		$c->addDescendingOrderByColumn(BatchJobLockPeer::URGENCY);
		$c->addDescendingOrderByColumn(BatchJobLockPeer::ESTIMATED_EFFORT);
		$c->setLimit($limit);
	
		$stmt = BatchJobLockPeer::doSelectStmt($c);
		$jobIds = $stmt->fetchAll(PDO::FETCH_COLUMN);
	
		// Suspend chosen ids
		$res = self::setJobsStatus($jobIds, BatchJob::BATCHJOB_STATUS_SUSPEND, BatchJob::BATCHJOB_STATUS_PENDING, true);
		$rootJobIds = self::suspendRootJobs($jobIds);
		self::moveToSuspendedJobsTable(array_merge($jobIds, $rootJobIds));
		KalturaLog::info("$res jobs of partner ($partnerId) job type ($jobType / $jobSubType) on DC ($dc) were suspended");
		KalturaLog::info("As a result, ". count($rootJobIds) . " root jobs were suspended.");
	}
	
	/**
	 * This function unsuspends up to ($limit) jobs of a given ($dc, $partnerId, $jobType, $jobSubType)
	 */
	private static function unsuspendJobs($limit, $dc, $partnerId, $jobType, $jobSubType) {
	
		$limit = min(self::MAX_PROCESSED_ROWS, $limit);
		
		// Find IDs from Batch Job Suspend Table
		$c = new Criteria();
		$c->addSelectColumn(BatchJobLockSuspendPeer::ID);
		$c->add( BatchJobLockSuspendPeer::DC, $dc, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::PARTNER_ID, $partnerId, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::JOB_TYPE, $jobType, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::JOB_SUB_TYPE, $jobSubType, Criteria::EQUAL);
		$c->add( BatchJobLockSuspendPeer::STATUS, array(BatchJob::BATCHJOB_STATUS_SUSPEND, BatchJob::BATCHJOB_STATUS_SUSPEND_ALMOST_DONE), Criteria::IN);
	
		$c->addAscendingOrderByColumn(BatchJobLockSuspendPeer::PRIORITY);
		$c->addAscendingOrderByColumn(BatchJobLockSuspendPeer::URGENCY);
		$c->addAscendingOrderByColumn(BatchJobLockSuspendPeer::ESTIMATED_EFFORT);
		$c->setLimit($limit);
	
		$stmt = BatchJobLockSuspendPeer::doSelectStmt($c);
		$jobIds = $stmt->fetchAll(PDO::FETCH_COLUMN);
	
		// Return the jobs from batch_job_lock_suspend table
		self::moveFromSuspendedJobsTable($jobIds);
		$rootJobIds = self::unsuspendRootJob($jobIds);
		// Update the jobs status to pending
		$res = self::setJobsStatus($jobIds, BatchJob::BATCHJOB_STATUS_PENDING, BatchJob::BATCHJOB_STATUS_SUSPEND, false);
		$resRoot = self::setJobsStatus($rootJobIds, BatchJob::BATCHJOB_STATUS_ALMOST_DONE, BatchJob::BATCHJOB_STATUS_SUSPEND_ALMOST_DONE, false);
		KalturaLog::info("$res jobs of partner ($partnerId) job type ($jobType / $jobSubType) on DC ($dc) were unsuspended");
		KalturaLog::info("As a result $resRoot were unsuspended.");
	}
	
	/**
	 * This function sets the status of the given $jobIds to the given $status.
	 * We're saved from the kFlowManager::updatedJob callback since we work directly through the DB. 
	 * No callbacks are called.
	 * Returns the number of the affected rows.
	 */
	private static function setJobsStatus($jobIds, $status, $oldStatus, $addSchedulerCond) {
	
		$suspenderUpdateChunk = self::getSuspenderUpdateChunk();
		
		$affectedRows = 0;
{'fraction_non_alphanumeric': 0.14807585052983827, 'fraction_numerical': 0.001742889012827663, 'mean_word_length': 5.541267669858641, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 95, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4325275', 'n_tokens_mistral': 5468, 'n_tokens_neox': 5122, 'n_words': 984}