<commit_before><?php
 
/*****************************
 * Includes & Globals
 */
ini_set("memory_limit","512M");

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	/****************************
	 * KChunkedEncodeMemcacheWrap
	 */
	class KChunkedEncodeMemcacheWrap implements KChunkedEncodeDistrExecInterface
	{
		const WRITE_IDX = "ChunkedEncodeWriteIdx";
		const READ_IDX  = "ChunkedEncodeReadIdx";
		
		const JOB_KEYNAME_PREFIX = "ChunkedEncodeJob";
		const SEMAPHORE_PREFIX = "ChunkedEncodeSemaphore";
		
		protected $cacheStore = null;
		protected $expiry = 86400;
		protected $memcacheConfig = null;
		
		protected $writeIndex = null;
		protected $readIndex = null;
				
		protected $storeToken = null;
		
		/* ---------------------------
		 * C'tor
		 */
		public function __construct($storeToken=null)
		{
			$this->storeToken = $storeToken;
		}

		/* ---------------------------
		 * Setup
		 */
		public function Setup($config)
		{
			$this->cacheStore = new kInfraMemcacheCacheWrapper();
			if($this->cacheStore->init($config)===false){
				KalturaLog::log("cacheStore failed to initialize with config:".serialize($config));
				false;
			}
			$this->memcacheConfig = $config;
			return $this->initialize();
		}
		
		/* ---------------------------
		 * Setup
		 */
		public function SetupWithCacheType($cacheType)
		{
			if(!isset($cacheType))
				$cacheType = kCacheManager::CACHE_TYPE_LIVE_MEDIA_SERVER . '_0';
				/*
				 * Create the memcacah store object
				 */
			$this->cacheStore = kCacheManager::getSingleLayerCache($cacheType);
			if(!$this->cacheStore) {
				KalturaLog::log("cacheStore is null. cacheType: $cacheType . returning false");
				return false;
			}
			
			return $this->initialize();
		}
		
		/* ---------------------------
		 * SaveJob
		 *	Store job to memcache storage
		 */
		public function SaveJob($job)
		{
			$key = $this->getJobKeyName($job->keyIdx);
			$str = serialize($job);
			if($this->set($key, $str, $this->expiry)===false){
				KalturaLog::log("Session($job->session) - Failed to set job $key($str)");
				return false;
			}
				// Just to remove non printables from the log msg
			$str = preg_replace('/[\x00-\x1F\x80-\xFF]/', '', $str);
			KalturaLog::log("Session($job->session) - Set job $key($str)");
			return true;
		}
		 
		/* ---------------------------
		 * AddJob
		 */
		public function AddJob($job)
		{
			$job->keyIdx = $this->allocateJobIndex();
			return $this->SaveJob($job);
		}
		
		/* ---------------------------
		 * FetchJob
		 *	Get job from memcache storage
		 */
		public function FetchJob($keyIdx, $maxTries=10)
		{
			$key = $this->getJobKeyName($keyIdx);
			/*
			 * 10 attempts to get the job object
			 */
			for($try=0; $try<$maxTries; $try++) {
				if(($jobStr=$this->get($key))!==false){
					break;
				}
				usleep(rand(0,30000));
				KalturaLog::log("Attempt($try) to fetch job ($keyIdx)");
			}
			if($try==$maxTries){
				KalturaLog::log("Job($keyIdx) - Failed to get job ($keyIdx)");
				return false;
			}
			$job = unserialize($jobStr);
			return $job;
		}
		 
		/* ---------------------------
		 * DeleteJob
		 */
		public function DeleteJob($keyIdx)
		{
			$key = $this->getJobKeyName($keyIdx);
			$this->delete($key);
			KalturaLog::log("Job($keyIdx) - Deleted");
			return true;
		}
		
		/* ---------------------------
		 * GetActiveSessions
		 *	
		 */
		public function GetActiveSessions()
		{
			if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
				KalturaLog::log("ERROR: Missing write or read index ");
				return false;
			}
				/*
				 * Stop fetching attempts if there are no unread objects
				 */
			if($readIndex>=$writeIndex+1) 
				return array();;

			KalturaLog::log("RD:$readIndex, WR:$writeIndex");
			$sessions = array();
			for($idx=$readIndex; $idx<$writeIndex; $idx++){
				$job = $this->FetchJob($idx);
				if($job===false){
					continue;
				}
				if(array_key_exists($job->session, $sessions)==true)
					$sessions[$job->session] = $sessions[$job->session]+1;
				else
					$sessions[$job->session] = 1;
			}
			return $sessions;
		}
		
		/* ---------------------------
		 * initilize
		 */
		protected function initialize()
		{
				/*
				 * Get the global /read/write indexes,
				 * if don't exist - create them w/out expiry time
				 */
			$this->fetchReadWriteIndexes($writeIndex, $readIndex);
			$writeIndexKeyName = $this->getWriteIndexKeyName();
			if(!isset($writeIndex)){
				$writeIndex = 0;
				if($this->set($writeIndexKeyName,$writeIndex,0)===false) {
					KalturaLog::log("Failed to create WR index ($writeIndexKeyName)");
					return false;
				}
			}
			$this->writeIndex = $writeIndex;
			KalturaLog::log("Current WR index value ($writeIndexKeyName:$writeIndex)");
			
			$readIndexKeyName = $this->getReadIndexKeyName();
			if(!isset($readIndex)){
				$readIndex = 1;
				if($this->set($readIndexKeyName,$readIndex,0)===false) {
					KalturaLog::log("Failed to create RD index ($readIndexKeyName)");
					return false;
				}
			}
			$this->readIndex = $readIndex;
			KalturaLog::log("Current RD index value ($readIndexKeyName:$readIndex)");
			return true;
		}
		
		/* ---------------------------
		 *
		 
		public function incrementReadIndex()
		{
			return $this->increment($this->getReadIndexKeyName());
		}
*/
		/* ---------------------------
		 * fetchReadWriteIndexes
		 *	
		 */
		public function fetchReadWriteIndexes(&$writeIndex, &$readIndex)
		{
			$writeIndexKeyName = $this->getWriteIndexKeyName();
			if(($writeIndex=$this->get($writeIndexKeyName))===false){
				KalturaLog::log("Missing WR index ($writeIndexKeyName)");
				$writeIndex = null;
			}
			else
				KalturaLog::log("Current WR index value ($writeIndexKeyName:$writeIndex)");

			$readIndexKeyName = $this->getReadIndexKeyName();		
			if(($readIndex=$this->get($readIndexKeyName))===false){
				KalturaLog::log("Missing RD index ($readIndexKeyName)");
				$readIndex = null;
			}
			else
				KalturaLog::log("Current RD index value ($readIndexKeyName: $readIndex)");
			if(isset($readIndex) && isset($writeIndex))
				return true;
			else
				return false;
		}
		
		/* ---------------------------
		 * getReadIndexKeyName
		 *	Compose global read-index
		 */
		protected function getReadIndexKeyName()
		{
			return self::READ_IDX.$this->storeToken;
		}

		/* ---------------------------
		 * getWriteIndexKeyName
		 * 	Compose global write-index
		 */
		protected function getWriteIndexKeyName()
		{
			return self::WRITE_IDX.$this->storeToken;
		}

		/* ---------------------------
		 * getJobKeyName
		 *	Compose job key name, 
		 *	using manager::JOB_KEYNAME_PREFIX const and 'storeToken'
		 */
		protected function getJobKeyName($keyIdx)
		{
			return self::JOB_KEYNAME_PREFIX.$this->storeToken."_$keyIdx";
		}

		/* ---------------------------
		 * getSemaphoreKeyName
		 *	Compose job semaphore key name, 
		 * 	using manager::SEMAPHORE_PREFIX const and 'storeToken'
		 */
		protected function getSemaphoreKeyName($keyIdx)
		{
			return self::SEMAPHORE_PREFIX.$this->storeToken."_$keyIdx";
		}
		
		/* ---------------------------
		 * allocateJobIndex
		 *	Allocate an index for a new job
		 */
		protected function allocateJobIndex()
		{
			return $this->increment($this->getWriteIndexKeyName());
		}

		/* ---------------------------
		 * lock
		 *	based on memcache 'add'  functionality  
		 */
		protected function lock($key, $val, $attempDuration=60, $attemptSleep=0.005, $expiry=3600) // expiry = 1hr
		{
			$waited=0;
	KalturaLog::log("key($key), val($val)");
			$attemptSleep*=1000000;
			do {
				$addLasted = microtime(true);
				$rv = $this->cacheStore->add($key, $val, $expiry);
				$addLasted = round((microtime(true)-$addLasted),3);
				if($rv===true){
					break;
				}
				$sleepLasted = microtime(true);
				usleep(rand(0,$attemptSleep));
				$sleepLasted = round((microtime(true)-$sleepLasted),5);
				$waited+=($addLasted+$sleepLasted);
				$attempDuration-= ($addLasted+$sleepLasted);
	KalturaLog::log("attempDuration($attempDuration)");
			}
			while($attempDuration>0);
			if($waited>0){
	KalturaLog::log("EXIT:key($key),val($val): rv($rv),waited($waited)");
			}
			else {
	KalturaLog::log("EXIT:key($key),val($val): rv($rv)");
			}
			return $rv;
		}

		/* ---------------------------
		 * delete
		 */
		protected function delete($key)
		{
			return $this->cacheStore->delete($key);
		}
		
		/* ---------------------------
		 * add
		 *	memcache low level 
		 */
		protected function add($key, $val, $expiry=86400)
		{
			return $this->cacheStore->add($key, $val, $expiry);
		}
		
		/* ---------------------------
		 * set
		 *	memcache low level 
		 */
		protected function set($key, $val, $expiry=86400)
		{
			return $this->cacheStore->set($key, $val, $expiry);
		}
		
		/* ---------------------------
		 * get
		 *	memcache low level 
		 */
		protected function get($key)
		{
			return $this->cacheStore->get($key);
		}
		
		/* ---------------------------
		 * increment
		 *	memcache low level 
		 */
		protected function increment($key, $value=1)
		{
			return $this->cacheStore->increment($key, $value);
		}
		
		/* ---------------------------
		 * decrement
		 *	memcache low level 
		 */
		protected function decrement($key, $value=1)
		{
			return $this->cacheStore->decrement($key, $value);
		}
		
		/* ---------------------------
		 * ExecuteSession
		 */
		public static function ExecuteSession($host, $port, $token, $concurrent, $concurrentMin, $sessionName, $cmdLine, $sharedChunkPath = null)
		{
			KalturaLog::log("host:$host, port:$port, token:$token, concurrent:$concurrent, concurrentMin:$concurrentMin, sessionName:$sessionName, cmdLine:$cmdLine, sharedChunkPath:$sharedChunkPath");
			$storeManager = new KChunkedEncodeMemcacheWrap($token);
				// 'flags=1' stands for 'compress stored data'
			$config = array('host'=>$host, 'port'=>$port, 'flags'=>1);
			$storeManager->Setup($config);
			
			$setup = new KChunkedEncodeSetup();
			$setup->concurrent = $concurrent;
			$setup->concurrentMin = $concurrentMin;
			$setup->cleanUp = 0;
			$setup->cmd = $cmdLine;
			$setup->sharedChunkPath = $sharedChunkPath;

			$session = new KChunkedEncodeSessionManager($setup, $storeManager, $sessionName);
			
			if(($rv=$session->Initialize())!=true) {
				$session->Report();
				return $rv;
			}
			$rv = $session->Generate();
			$session->Report();
			return $rv;
		}

	}
	/*****************************
	 * End of KChunkedEncodeMemcacheWrap
	 *****************************/
	
	/****************************
	 * KChunkedEncodeMemcacheScheduler
	 */
	class KChunkedEncodeMemcacheScheduler extends KChunkedEncodeMemcacheWrap implements KChunkedEncodeDistrSchedInterface
	{
		protected $tmpFolder = null;
		
		/* ---------------------------
		 * C'tor
		 */
		public function __construct($storeToken=null, $tmpFolder=null)
		{
			parent::__construct($storeToken);
			$this->tmpFolder = $tmpFolder;
		}
		
		/* ---------------------------
		 * FetchNextJob
		 *	Get next job from the mmecache storage
		 * 	if missing => return false 
		 * fetchRangeRandMax - represents the max randomization range of the fetched job index (between the readIdx and writeIdx).
		 * Motivation - to ease race condition over the locking of the fetch/read idx, 
		 * when large number of srvs attempt to fetch the same job. 
		 * This situation occurred in AWS env, with high connect/'walk around' times to DC memcache srv
		 */
		public function FetchNextJob($fetchRangeRandMax=0){
			if($fetchRangeRandMax==0)
				return $this->FetchNextJobIndexInc($fetchRangeRandMax);
			else
				return $this->FetchNextJobRandom($fetchRangeRandMax);
		}
		public function FetchNextJobRandom($fetchRangeRandMax=50)
		{
			$writeIndex = null;
			$readIndex = null;
			
			// semaphore token - process-id + hostname + rand
			$semaphoreToken = getmypid().".".gethostname().".".rand();
			while(true) {
				if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
					KalturaLog::log("ERROR: Missing write or read index ");
					return false;
				}
					/*
					 * Stop fetching if there are no unread objects
					 */
				if($readIndex>=$writeIndex+1) 
					break;
				
					/*
					 * Evaluate the fetch job index, within the 'fetchRangeRandMax' value.
					 * If the chunk Q is smaller than the fetch range - take the job in the readIdx
					 */
				$queueSize = $writeIndex-$readIndex;
				if($fetchRangeRandMax==0 || $queueSize<$fetchRangeRandMax)
					$fetchIndex = $readIndex;
				else
					$fetchIndex = rand($readIndex, min($readIndex+$fetchRangeRandMax, $writeIndex));
                			KalturaLog::log("RD:$readIndex, WR:$writeIndex, FCH:$fetchIndex");

					/*
					 * Try to lock the next unread job object
					 * if failed - carry on to next
					 */
				$semaphoreKey = $this->getSemaphoreKeyName($fetchIndex);
				
				$rv = $this->lock($semaphoreKey, $semaphoreToken, 0);
				if($rv!==true){
					if($fetchIndex==$readIndex && $this->setReadIndex($readIndex+1)===false)
						return false;
					KalturaLog::log("Unable to lock fetchIndex($fetchIndex), skipping to the next ");
					continue;
				}
				
					/*
					 * Try to fetch the job object from the memcache storage
					 * If failed - delete the sempahore (unlock) and try the next one
					 */
				$job = $this->FetchJob($fetchIndex); 
				if($job!==false) {
					return $job;
				}
				else {
					$this->delete($semaphoreKey);
					KalturaLog::log("Unable to access job ($fetchIndex), skip to next");
					if($fetchIndex==$readIndex && $this->setReadIndex($readIndex+1)===false)
						return false;
				}
			}
			
			return null;
		}
		
		public function FetchNextJobIndexInc($fetchRangeRandMax=50){
			if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
				KalturaLog::log("ERROR: Missing write or read index ");
				return false;
			}
				/*
				 * Stop fetching if there are no unread objects
				 */
			if($readIndex>$writeIndex){
				return null;
			}
			
			$fetchIndex = $this->incrementReadIndex()-1;
			KalturaLog::log("fetchIndex:$fetchIndex (new)");

			$semaphoreKey = $this->getSemaphoreKeyName($fetchIndex);
			$semaphoreToken = getmypid().".".gethostname().".".rand();
			$rvLock = null;
			$attempts=10;
			while(true) {
						/*
						 * Try to lock the next unread job object
						 * if failed - carry on to next
						 */
				if(is_null($rvLock)) $rvLock = $this->lock($semaphoreKey, $semaphoreToken, 0);
				if($rvLock!==true){
					KalturaLog::log("Unable to lock fetchIndex($fetchIndex), retry");
				}
				else {
						/*
						 * Try to fetch the job object from the memcache
						 */
					$job = $this->FetchJob($fetchIndex); 
					if($job!==false) {
						return $job;
					}
				}

				if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
					KalturaLog::log("ERROR: Missing write or read index ");
					break;
				}
				
				if($fetchIndex<$readIndex-1000 || $attempts==0) {
					KalturaLog::log("Unable to access job($fetchIndex), while rdIdx($readIndex) moved on, skipping");
					break;
				}

				$attempts--;
				KalturaLog::log("Unable to access job ($fetchIndex), retry ($attempts)");
				usleep(rand(0,100000));
			}
			
			$this->delete($semaphoreKey);
			return false;
		}

		/* ---------------------------
		 * GetRunningJobs
		 */
		public function GetRunningJobs($lookBackward=500) {
			KalturaLog::log("lookBackward:$lookBackward");
			if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
				KalturaLog::log("ERROR: Missing write or read index ");
				return false;
			}
			$localHostname = gethostname();
			$jobs = array();
			$idx=max(0,$readIndex-$lookBackward);
			$idxMax = min($readIndex+$lookBackward,$writeIndex);
			for(; $idx<$idxMax; $idx++) {
				$job = $this->FetchJob($idx,1);
				if($job!==false && $job->state==$job::STATE_RUNNING 
				&& strcmp($localHostname, $job->hostname)=== 0 && KProcessExecutionData::isProcessRunning($job->process)) {
					$jobs[] = $job;
					KalturaLog::log("Job: keyIdx($idx), pId($job->process)");
				}
			}
			KalturaLog::log("Found running jobs - ".count($jobs));
			return($jobs);
		}
		
		/* ---------------------------
		 * RefreshJobs
		 */
		public function RefreshJobs($maxSlots, $fetchRangeRandMax, &$jobs)
		{
				/*
				 * Get list of per scheduler running jobs
				 */
			$refreshed = $this->refetchJobs($jobs);
			$running = count($jobs);
			if($running>0 && $refreshed==0){
				return false;
			}
				/*
				 * If there are no free execution slots - wait and retry
				 */
			if($running>=$maxSlots) {
				KalturaLog::log("Running:$running - No free job slots, maxSlots:$maxSlots");
				return null;
			}

				/*
				 * If there are no pending jobs - wait and retry
				 */
			$job = $this->FetchNextJob($fetchRangeRandMax);
			if($job===null){
				KalturaLog::log("Running:$running - No pending jobs");
				return null;
			}
			else if($job===false){
				KalturaLog::log("Failed to fetch next job");
				return false;
			}

			if($this->ExecuteJob($job)==true)
				$jobs[$job->keyIdx] = $job;
			
			return true;
		}

		/* ---------------------------
		 * refetchJobs
		 * 	Reload the givven jobs array from memcache storage
		 * 	Remove finished or retried jobs 
		 */
		protected function refetchJobs(&$jobs)
		{
			$cnt=0;
			foreach($jobs as $idx=>$job) {
				$job = $this->FetchJob($job->keyIdx);
				if($job===false) {
					KalturaLog::log("Missing $idx");
					continue;
				}
				if($job->isFinished() || $job->isRetry()) {
					unset($jobs[$idx]);
				}
				else {
					$jobs[$idx] = $job;
				}
				$cnt++;
			}
			return $cnt;
		}

		/* ---------------------------
		 * getPhpChunkJobLogName
		 */
		protected static function getPhpChunkJobLogName($job)
		{
			return("$job->session"."_$job->id"."_$job->keyIdx".".log");
		}
		
		/* ---------------------------
		 *
		 */
		public function ExecuteJob($job)
		{
			$job->queueTime = time();
			$this->SaveJob($job);

			if(!isset($this->tmpFolder))
				$this->tmpFolder = "/tmp";			
			
			$logFolder = $this->tmpFolder;
			
			$logName = "$logFolder/".$this->getPhpChunkJobLogName($job);
			{
				$cmdLine = 'php -r "';
				$cmdLine.= 'require_once \'/opt/kaltura/app/batch/bootstrap.php\';';
///////////////
// DEBUG ONLY
// $dirName = dirname(__FILE__); //"/web2/content/shared/tmp/qualityTest/TestBench.11";
// $cmdLine.= 'require_once \''.$dirName.'/KChunkedEncodeUtils.php\';';
// $cmdLine.= 'require_once \''.$dirName.'/KChunkedEncodeMemcacheWrap.php\';';
//$cmdLine.= 'require_once \''.$dirName.'/KFFMpegMediaParser.php\';';
///////////////
				$cmdLine.= '\$rv=KChunkedEncodeMemcacheScheduler::ExecuteJobCommand(';
				$cmdLine.= '\''.($this->memcacheConfig['host']).'\',';
				$cmdLine.= '\''.($this->memcacheConfig['port']).'\',';
				$cmdLine.= '\''.($this->storeToken).'\',';
				$cmdLine.= $job->keyIdx.',';
				$cmdLine.= '\''.($this->tmpFolder).'\');';
				$cmdLine.= 'if(\$rv==false) exit(1);';
				$cmdLine.= '"';
			}
			$chunk_job_pid_file = $this->tmpFolder."/chunk_job_pid_".$job->session."_".$job->keyIdx.".log";
			$cmdLine.= " > $logName 2>&1 & echo $! > $chunk_job_pid_file";

			KalturaLog::log($cmdLine);

			$output = system($cmdLine, $rv);
			if($rv!=0) {
				$job->state = $job::STATE_FAIL;
				$this->SaveJob($job);
			}
			else {
				$job->process = (int)kFile::getFileContent($chunk_job_pid_file);
				kFile::unlink($chunk_job_pid_file);
			}
			KalturaLog::log("id:$job->id,keyIdx:$job->keyIdx,rv:$rv,process:$job->process,cmdLine:$cmdLine");
			return true;
		}
		
		/**
		 * This method executed one job fetched from the chunk encoding queue
		 * Please note that changing this methods signature may require adjusting the regex in KillJobsCommand method
		 *
		 * @param $host
		 * @param $port
		 * @param $token
		 * @param $jobIndex
		 * @param string $tmpPromptFolder
		 * @return bool
		 */
		public static function ExecuteJobCommand($host, $port, $token, $jobIndex, $tmpPromptFolder="/tmp")
		{
			KalturaLog::log("host:$host, port:$port, token:$token, jobIndex:$jobIndex");
			$storeManager = new KChunkedEncodeMemcacheWrap($token);
				// 'flags=1' stands for 'compress stored data'
			$config = array('host'=>$host, 'port'=>$port, 'flags'=>1);
			$storeManager->Setup($config);
			
			if(!isset($jobIndex)) {
				return false;
			}

			$job = $storeManager->FetchJob($jobIndex);
			if($job===false)
				return false;
			
			kBatchUtils::tryLoadKconfConfig();
			
			$job->startTime = time();
			$job->process = getmypid();
			$job->hostname = gethostname();
			$job->state = $job::STATE_RUNNING;
			$storeManager->SaveJob($job);
			
			$outFilename = null;
			if(is_array($job->cmdLine)) {
				$cmdLine = $job->cmdLine[0];
				$outFilenames = isset($job->cmdLine[1]) ? $job->cmdLine[1] : null;;
				$outFilename = is_array($outFilenames) ? $outFilenames[0] : $outFilenames;
				//Added to support use cases where the shared file system is not NFS but rather a remote object storage.
				$sharedChunkPaths = isset($job->cmdLine[2]) ? $job->cmdLine[2] : null;
//				$outFilename = $job->cmdLine[1];
					/*
					 * Switch chunk generation to local disk folder (tmpPromptFolder)
					 * - replace the out file in tha cmdline to '/tmp' flolder
					 * - move all generated files back to the 'normal' tmp path
					 */
				$outFilenameInfo = null;
					// !!!TO DISABLE chunk-to-tmp flow, turn the 'true' to 'false' in condition bellow!!!
				if(isset($tmpPromptFolder) && true) {
					$outFilenameInfo = pathinfo($outFilename);
					$outFilename=$tmpPromptFolder.'/'.$outFilenameInfo['basename'];
					$cmdLine=str_replace($outFilenameInfo['dirname'], $tmpPromptFolder, $cmdLine);
						//Adjust the 'shared' mode outputs to 'chk-to-tmp' flow
					if(is_array($outFilenames)) {
						$storageFilenames = array();
						foreach($outFilenames as $idx=>$tmpName) {
							$storageFilenames[$idx] = $outFilenames[$idx];
							$tmpNameInfo = pathinfo($tmpName);
							$outFilenames[$idx]=$tmpPromptFolder.'/'.$tmpNameInfo['basename'];
						}
					}
				}
			}
			else
				$cmdLine = $job->cmdLine;
				/*
				 * Up to 2 attempts for chunk generation, in order to overcome possible
				 * zeroed frames issue, caused by too large GOPs in the source files 
				 */
			for($i=0;$i<2;$i++) {
				if($i>0){
					$cmdLine = self::fixCmdLineBackOffset($cmdLine);
					KalturaLog::log("2nd attempt, due to empty frames, fixed cmdLine:$cmdLine");
					sleep(3);
				}
				exec($cmdLine,$op,$rv);
				$job->finishTime = time();
				if($rv!=0) {
					$job->state = $job::STATE_FAIL;
					$rvStr = "FAILED - rv($rv),";
					break;
				}
					// No need to check for empties for 1st chunk and audio chunks
				if($job->id==0 || strstr($outFilename,'.vid')===false)
					break;
				if(KFFMpegMediaParser::detectEmptyFrames("ffmpeg", "ffprobe", $outFilename)===false)
					break;
			}

			if($rv==0) {
				if(isset($outFilename) && strstr($outFilename,'.vid')) {
					$stat = new KChunkFramesStat();
						// VP9 / AV1 requires MP4 file container, they don't comply w/MPEGTS
						// X264 / X265 comply w/both MP4 & MPEGTS, but meanwhile we'll continue w/MPEGTS
					if((strstr($cmdLine,"libvpx-vp9")!==false) || (strstr($cmdLine,"libaom-av1")!==false))
						$rv = $stat->getDataMP4($outFilename,"ffprobe","ffmpeg",$tmpPromptFolder);
					else
						$rv = $stat->getDataMpegts($outFilename,"ffprobe","ffmpeg",$tmpPromptFolder);
					$job->stat = $stat;
				}

				if(isset($stat) && $stat->isEmpty()){
					$job->state = $job::STATE_FAIL;
					$rvStr = "FAILED - missing chunk stat,";
					$job->msg = "missing chunk stat";
				}
				else {
					$job->state = $job::STATE_SUCCESS;
					$rvStr = "SUCCESS -";
				}<commit_msg>pass ffmpegBin/ffprobeBin to ExecuteJobCommand func<commit_after><?php
 
/*****************************
 * Includes & Globals
 */
ini_set("memory_limit","512M");

/////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

	/****************************
	 * KChunkedEncodeMemcacheWrap
	 */
	class KChunkedEncodeMemcacheWrap implements KChunkedEncodeDistrExecInterface
	{
		const WRITE_IDX = "ChunkedEncodeWriteIdx";
		const READ_IDX  = "ChunkedEncodeReadIdx";
		
		const JOB_KEYNAME_PREFIX = "ChunkedEncodeJob";
		const SEMAPHORE_PREFIX = "ChunkedEncodeSemaphore";
		
		protected $cacheStore = null;
		protected $expiry = 86400;
		protected $memcacheConfig = null;
		
		protected $writeIndex = null;
		protected $readIndex = null;
				
		protected $storeToken = null;
		
		/* ---------------------------
		 * C'tor
		 */
		public function __construct($storeToken=null)
		{
			$this->storeToken = $storeToken;
		}

		/* ---------------------------
		 * Setup
		 */
		public function Setup($config)
		{
			$this->cacheStore = new kInfraMemcacheCacheWrapper();
			if($this->cacheStore->init($config)===false){
				KalturaLog::log("cacheStore failed to initialize with config:".serialize($config));
				false;
			}
			$this->memcacheConfig = $config;
			return $this->initialize();
		}
		
		/* ---------------------------
		 * Setup
		 */
		public function SetupWithCacheType($cacheType)
		{
			if(!isset($cacheType))
				$cacheType = kCacheManager::CACHE_TYPE_LIVE_MEDIA_SERVER . '_0';
				/*
				 * Create the memcacah store object
				 */
			$this->cacheStore = kCacheManager::getSingleLayerCache($cacheType);
			if(!$this->cacheStore) {
				KalturaLog::log("cacheStore is null. cacheType: $cacheType . returning false");
				return false;
			}
			
			return $this->initialize();
		}
		
		/* ---------------------------
		 * SaveJob
		 *	Store job to memcache storage
		 */
		public function SaveJob($job)
		{
			$key = $this->getJobKeyName($job->keyIdx);
			$str = serialize($job);
			if($this->set($key, $str, $this->expiry)===false){
				KalturaLog::log("Session($job->session) - Failed to set job $key($str)");
				return false;
			}
				// Just to remove non printables from the log msg
			$str = preg_replace('/[\x00-\x1F\x80-\xFF]/', '', $str);
			KalturaLog::log("Session($job->session) - Set job $key($str)");
			return true;
		}
		 
		/* ---------------------------
		 * AddJob
		 */
		public function AddJob($job)
		{
			$job->keyIdx = $this->allocateJobIndex();
			return $this->SaveJob($job);
		}
		
		/* ---------------------------
		 * FetchJob
		 *	Get job from memcache storage
		 */
		public function FetchJob($keyIdx, $maxTries=10)
		{
			$key = $this->getJobKeyName($keyIdx);
			/*
			 * 10 attempts to get the job object
			 */
			for($try=0; $try<$maxTries; $try++) {
				if(($jobStr=$this->get($key))!==false){
					break;
				}
				usleep(rand(0,30000));
				KalturaLog::log("Attempt($try) to fetch job ($keyIdx)");
			}
			if($try==$maxTries){
				KalturaLog::log("Job($keyIdx) - Failed to get job ($keyIdx)");
				return false;
			}
			$job = unserialize($jobStr);
			return $job;
		}
		 
		/* ---------------------------
		 * DeleteJob
		 */
		public function DeleteJob($keyIdx)
		{
			$key = $this->getJobKeyName($keyIdx);
			$this->delete($key);
			KalturaLog::log("Job($keyIdx) - Deleted");
			return true;
		}
		
		/* ---------------------------
		 * GetActiveSessions
		 *	
		 */
		public function GetActiveSessions()
		{
			if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
				KalturaLog::log("ERROR: Missing write or read index ");
				return false;
			}
				/*
				 * Stop fetching attempts if there are no unread objects
				 */
			if($readIndex>=$writeIndex+1) 
				return array();;

			KalturaLog::log("RD:$readIndex, WR:$writeIndex");
			$sessions = array();
			for($idx=$readIndex; $idx<$writeIndex; $idx++){
				$job = $this->FetchJob($idx);
				if($job===false){
					continue;
				}
				if(array_key_exists($job->session, $sessions)==true)
					$sessions[$job->session] = $sessions[$job->session]+1;
				else
					$sessions[$job->session] = 1;
			}
			return $sessions;
		}
		
		/* ---------------------------
		 * initilize
		 */
		protected function initialize()
		{
				/*
				 * Get the global /read/write indexes,
				 * if don't exist - create them w/out expiry time
				 */
			$this->fetchReadWriteIndexes($writeIndex, $readIndex);
			$writeIndexKeyName = $this->getWriteIndexKeyName();
			if(!isset($writeIndex)){
				$writeIndex = 0;
				if($this->set($writeIndexKeyName,$writeIndex,0)===false) {
					KalturaLog::log("Failed to create WR index ($writeIndexKeyName)");
					return false;
				}
			}
			$this->writeIndex = $writeIndex;
			KalturaLog::log("Current WR index value ($writeIndexKeyName:$writeIndex)");
			
			$readIndexKeyName = $this->getReadIndexKeyName();
			if(!isset($readIndex)){
				$readIndex = 1;
				if($this->set($readIndexKeyName,$readIndex,0)===false) {
					KalturaLog::log("Failed to create RD index ($readIndexKeyName)");
					return false;
				}
			}
			$this->readIndex = $readIndex;
			KalturaLog::log("Current RD index value ($readIndexKeyName:$readIndex)");
			return true;
		}
		
		/* ---------------------------
		 *
		 
		public function incrementReadIndex()
		{
			return $this->increment($this->getReadIndexKeyName());
		}
*/
		/* ---------------------------
		 * fetchReadWriteIndexes
		 *	
		 */
		public function fetchReadWriteIndexes(&$writeIndex, &$readIndex)
		{
			$writeIndexKeyName = $this->getWriteIndexKeyName();
			if(($writeIndex=$this->get($writeIndexKeyName))===false){
				KalturaLog::log("Missing WR index ($writeIndexKeyName)");
				$writeIndex = null;
			}
			else
				KalturaLog::log("Current WR index value ($writeIndexKeyName:$writeIndex)");

			$readIndexKeyName = $this->getReadIndexKeyName();		
			if(($readIndex=$this->get($readIndexKeyName))===false){
				KalturaLog::log("Missing RD index ($readIndexKeyName)");
				$readIndex = null;
			}
			else
				KalturaLog::log("Current RD index value ($readIndexKeyName: $readIndex)");
			if(isset($readIndex) && isset($writeIndex))
				return true;
			else
				return false;
		}
		
		/* ---------------------------
		 * getReadIndexKeyName
		 *	Compose global read-index
		 */
		protected function getReadIndexKeyName()
		{
			return self::READ_IDX.$this->storeToken;
		}

		/* ---------------------------
		 * getWriteIndexKeyName
		 * 	Compose global write-index
		 */
		protected function getWriteIndexKeyName()
		{
			return self::WRITE_IDX.$this->storeToken;
		}

		/* ---------------------------
		 * getJobKeyName
		 *	Compose job key name, 
		 *	using manager::JOB_KEYNAME_PREFIX const and 'storeToken'
		 */
		protected function getJobKeyName($keyIdx)
		{
			return self::JOB_KEYNAME_PREFIX.$this->storeToken."_$keyIdx";
		}

		/* ---------------------------
		 * getSemaphoreKeyName
		 *	Compose job semaphore key name, 
		 * 	using manager::SEMAPHORE_PREFIX const and 'storeToken'
		 */
		protected function getSemaphoreKeyName($keyIdx)
		{
			return self::SEMAPHORE_PREFIX.$this->storeToken."_$keyIdx";
		}
		
		/* ---------------------------
		 * allocateJobIndex
		 *	Allocate an index for a new job
		 */
		protected function allocateJobIndex()
		{
			return $this->increment($this->getWriteIndexKeyName());
		}

		/* ---------------------------
		 * lock
		 *	based on memcache 'add'  functionality  
		 */
		protected function lock($key, $val, $attempDuration=60, $attemptSleep=0.005, $expiry=3600) // expiry = 1hr
		{
			$waited=0;
	KalturaLog::log("key($key), val($val)");
			$attemptSleep*=1000000;
			do {
				$addLasted = microtime(true);
				$rv = $this->cacheStore->add($key, $val, $expiry);
				$addLasted = round((microtime(true)-$addLasted),3);
				if($rv===true){
					break;
				}
				$sleepLasted = microtime(true);
				usleep(rand(0,$attemptSleep));
				$sleepLasted = round((microtime(true)-$sleepLasted),5);
				$waited+=($addLasted+$sleepLasted);
				$attempDuration-= ($addLasted+$sleepLasted);
	KalturaLog::log("attempDuration($attempDuration)");
			}
			while($attempDuration>0);
			if($waited>0){
	KalturaLog::log("EXIT:key($key),val($val): rv($rv),waited($waited)");
			}
			else {
	KalturaLog::log("EXIT:key($key),val($val): rv($rv)");
			}
			return $rv;
		}

		/* ---------------------------
		 * delete
		 */
		protected function delete($key)
		{
			return $this->cacheStore->delete($key);
		}
		
		/* ---------------------------
		 * add
		 *	memcache low level 
		 */
		protected function add($key, $val, $expiry=86400)
		{
			return $this->cacheStore->add($key, $val, $expiry);
		}
		
		/* ---------------------------
		 * set
		 *	memcache low level 
		 */
		protected function set($key, $val, $expiry=86400)
		{
			return $this->cacheStore->set($key, $val, $expiry);
		}
		
		/* ---------------------------
		 * get
		 *	memcache low level 
		 */
		protected function get($key)
		{
			return $this->cacheStore->get($key);
		}
		
		/* ---------------------------
		 * increment
		 *	memcache low level 
		 */
		protected function increment($key, $value=1)
		{
			return $this->cacheStore->increment($key, $value);
		}
		
		/* ---------------------------
		 * decrement
		 *	memcache low level 
		 */
		protected function decrement($key, $value=1)
		{
			return $this->cacheStore->decrement($key, $value);
		}
		
		/* ---------------------------
		 * ExecuteSession
		 */
		public static function ExecuteSession($host, $port, $token, $concurrent, $concurrentMin, $sessionName, $cmdLine, $sharedChunkPath = null)
		{
			KalturaLog::log("host:$host, port:$port, token:$token, concurrent:$concurrent, concurrentMin:$concurrentMin, sessionName:$sessionName, cmdLine:$cmdLine, sharedChunkPath:$sharedChunkPath");
			$storeManager = new KChunkedEncodeMemcacheWrap($token);
				// 'flags=1' stands for 'compress stored data'
			$config = array('host'=>$host, 'port'=>$port, 'flags'=>1);
			$storeManager->Setup($config);
			
			$setup = new KChunkedEncodeSetup();
			$setup->concurrent = $concurrent;
			$setup->concurrentMin = $concurrentMin;
			$setup->cleanUp = 0;
			$setup->cmd = $cmdLine;
			$setup->sharedChunkPath = $sharedChunkPath;

			$session = new KChunkedEncodeSessionManager($setup, $storeManager, $sessionName);
			
			if(($rv=$session->Initialize())!=true) {
				$session->Report();
				return $rv;
			}
			$rv = $session->Generate();
			$session->Report();
			return $rv;
		}

	}
	/*****************************
	 * End of KChunkedEncodeMemcacheWrap
	 *****************************/
	
	/****************************
	 * KChunkedEncodeMemcacheScheduler
	 */
	class KChunkedEncodeMemcacheScheduler extends KChunkedEncodeMemcacheWrap implements KChunkedEncodeDistrSchedInterface
	{
		protected $tmpFolder = null;
		
		/* ---------------------------
		 * C'tor
		 */
		public function __construct($storeToken=null, $tmpFolder=null)
		{
			parent::__construct($storeToken);
			$this->tmpFolder = $tmpFolder;
		}
		
		/* ---------------------------
		 * FetchNextJob
		 *	Get next job from the mmecache storage
		 * 	if missing => return false 
		 * fetchRangeRandMax - represents the max randomization range of the fetched job index (between the readIdx and writeIdx).
		 * Motivation - to ease race condition over the locking of the fetch/read idx, 
		 * when large number of srvs attempt to fetch the same job. 
		 * This situation occurred in AWS env, with high connect/'walk around' times to DC memcache srv
		 */
		public function FetchNextJob($fetchRangeRandMax=0){
			if($fetchRangeRandMax==0)
				return $this->FetchNextJobIndexInc($fetchRangeRandMax);
			else
				return $this->FetchNextJobRandom($fetchRangeRandMax);
		}
		public function FetchNextJobRandom($fetchRangeRandMax=50)
		{
			$writeIndex = null;
			$readIndex = null;
			
			// semaphore token - process-id + hostname + rand
			$semaphoreToken = getmypid().".".gethostname().".".rand();
			while(true) {
				if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
					KalturaLog::log("ERROR: Missing write or read index ");
					return false;
				}
					/*
					 * Stop fetching if there are no unread objects
					 */
				if($readIndex>=$writeIndex+1) 
					break;
				
					/*
					 * Evaluate the fetch job index, within the 'fetchRangeRandMax' value.
					 * If the chunk Q is smaller than the fetch range - take the job in the readIdx
					 */
				$queueSize = $writeIndex-$readIndex;
				if($fetchRangeRandMax==0 || $queueSize<$fetchRangeRandMax)
					$fetchIndex = $readIndex;
				else
					$fetchIndex = rand($readIndex, min($readIndex+$fetchRangeRandMax, $writeIndex));
                			KalturaLog::log("RD:$readIndex, WR:$writeIndex, FCH:$fetchIndex");

					/*
					 * Try to lock the next unread job object
					 * if failed - carry on to next
					 */
				$semaphoreKey = $this->getSemaphoreKeyName($fetchIndex);
				
				$rv = $this->lock($semaphoreKey, $semaphoreToken, 0);
				if($rv!==true){
					if($fetchIndex==$readIndex && $this->setReadIndex($readIndex+1)===false)
						return false;
					KalturaLog::log("Unable to lock fetchIndex($fetchIndex), skipping to the next ");
					continue;
				}
				
					/*
					 * Try to fetch the job object from the memcache storage
					 * If failed - delete the sempahore (unlock) and try the next one
					 */
				$job = $this->FetchJob($fetchIndex); 
				if($job!==false) {
					return $job;
				}
				else {
					$this->delete($semaphoreKey);
					KalturaLog::log("Unable to access job ($fetchIndex), skip to next");
					if($fetchIndex==$readIndex && $this->setReadIndex($readIndex+1)===false)
						return false;
				}
			}
			
			return null;
		}
		
		public function FetchNextJobIndexInc($fetchRangeRandMax=50){
			if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
				KalturaLog::log("ERROR: Missing write or read index ");
				return false;
			}
				/*
				 * Stop fetching if there are no unread objects
				 */
			if($readIndex>$writeIndex){
				return null;
			}
			
			$fetchIndex = $this->incrementReadIndex()-1;
			KalturaLog::log("fetchIndex:$fetchIndex (new)");

			$semaphoreKey = $this->getSemaphoreKeyName($fetchIndex);
			$semaphoreToken = getmypid().".".gethostname().".".rand();
			$rvLock = null;
			$attempts=10;
			while(true) {
						/*
						 * Try to lock the next unread job object
						 * if failed - carry on to next
						 */
				if(is_null($rvLock)) $rvLock = $this->lock($semaphoreKey, $semaphoreToken, 0);
				if($rvLock!==true){
					KalturaLog::log("Unable to lock fetchIndex($fetchIndex), retry");
				}
				else {
						/*
						 * Try to fetch the job object from the memcache
						 */
					$job = $this->FetchJob($fetchIndex); 
					if($job!==false) {
						return $job;
					}
				}

				if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
					KalturaLog::log("ERROR: Missing write or read index ");
					break;
				}
				
				if($fetchIndex<$readIndex-1000 || $attempts==0) {
					KalturaLog::log("Unable to access job($fetchIndex), while rdIdx($readIndex) moved on, skipping");
					break;
				}

				$attempts--;
				KalturaLog::log("Unable to access job ($fetchIndex), retry ($attempts)");
				usleep(rand(0,100000));
			}
			
			$this->delete($semaphoreKey);
			return false;
		}

		/* ---------------------------
		 * GetRunningJobs
		 */
		public function GetRunningJobs($lookBackward=500) {
			KalturaLog::log("lookBackward:$lookBackward");
			if($this->fetchReadWriteIndexes($writeIndex, $readIndex)===false){
				KalturaLog::log("ERROR: Missing write or read index ");
				return false;
			}
			$localHostname = gethostname();
			$jobs = array();
			$idx=max(0,$readIndex-$lookBackward);
			$idxMax = min($readIndex+$lookBackward,$writeIndex);
			for(; $idx<$idxMax; $idx++) {
				$job = $this->FetchJob($idx,1);
				if($job!==false && $job->state==$job::STATE_RUNNING 
				&& strcmp($localHostname, $job->hostname)=== 0 && KProcessExecutionData::isProcessRunning($job->process)) {
					$jobs[] = $job;
					KalturaLog::log("Job: keyIdx($idx), pId($job->process)");
				}
			}
			KalturaLog::log("Found running jobs - ".count($jobs));
			return($jobs);
		}
		
		/* ---------------------------
		 * RefreshJobs
		 */
		public function RefreshJobs($maxSlots, $fetchRangeRandMax, &$jobs)
		{
				/*
				 * Get list of per scheduler running jobs
				 */
			$refreshed = $this->refetchJobs($jobs);
			$running = count($jobs);
			if($running>0 && $refreshed==0){
				return false;
			}
				/*
				 * If there are no free execution slots - wait and retry
				 */
			if($running>=$maxSlots) {
				KalturaLog::log("Running:$running - No free job slots, maxSlots:$maxSlots");
				return null;
			}

				/*
				 * If there are no pending jobs - wait and retry
				 */
			$job = $this->FetchNextJob($fetchRangeRandMax);
			if($job===null){
				KalturaLog::log("Running:$running - No pending jobs");
				return null;
			}
			else if($job===false){
				KalturaLog::log("Failed to fetch next job");
				return false;
			}

			if($this->ExecuteJob($job)==true)
				$jobs[$job->keyIdx] = $job;
			
			return true;
		}

		/* ---------------------------
		 * refetchJobs
		 * 	Reload the givven jobs array from memcache storage
		 * 	Remove finished or retried jobs 
		 */
		protected function refetchJobs(&$jobs)
		{
			$cnt=0;
			foreach($jobs as $idx=>$job) {
				$job = $this->FetchJob($job->keyIdx);
				if($job===false) {
					KalturaLog::log("Missing $idx");
					continue;
				}
				if($job->isFinished() || $job->isRetry()) {
					unset($jobs[$idx]);
				}
				else {
					$jobs[$idx] = $job;
				}
				$cnt++;
			}
			return $cnt;
		}

		/* ---------------------------
		 * getPhpChunkJobLogName
		 */
		protected static function getPhpChunkJobLogName($job)
		{
			return("$job->session"."_$job->id"."_$job->keyIdx".".log");
		}
		
		/* ---------------------------
		 *
		 */
		public function ExecuteJob($job)
		{
			$job->queueTime = time();
			$this->SaveJob($job);

			if(!isset($this->tmpFolder))
				$this->tmpFolder = "/tmp";			
			
			$logFolder = $this->tmpFolder;
			
			$logName = "$logFolder/".$this->getPhpChunkJobLogName($job);
			{
				$cmdLine = 'php -r "';
				$cmdLine.= 'require_once \'/opt/kaltura/app/batch/bootstrap.php\';';
///////////////
// DEBUG ONLY
// $dirName = dirname(__FILE__); //"/web2/content/shared/tmp/qualityTest/TestBench.11";
// $cmdLine.= 'require_once \''.$dirName.'/KChunkedEncodeUtils.php\';';
// $cmdLine.= 'require_once \''.$dirName.'/KChunkedEncodeMemcacheWrap.php\';';
//$cmdLine.= 'require_once \''.$dirName.'/KFFMpegMediaParser.php\';';
///////////////
				$cmdLine.= '\$rv=KChunkedEncodeMemcacheScheduler::ExecuteJobCommand(';
				$cmdLine.= '\''.($this->memcacheConfig['host']).'\',';
				$cmdLine.= '\''.($this->memcacheConfig['port']).'\',';
				$cmdLine.= '\''.($this->storeToken).'\',';
				$cmdLine.= $job->keyIdx.',';
				$cmdLine.= '\''.($this->tmpFolder).'\');';
				$cmdLine.= 'if(\$rv==false) exit(1);';
				$cmdLine.= '"';
			}
			$chunk_job_pid_file = $this->tmpFolder."/chunk_job_pid_".$job->session."_".$job->keyIdx.".log";
			$cmdLine.= " > $logName 2>&1 & echo $! > $chunk_job_pid_file";

			KalturaLog::log($cmdLine);

			$output = system($cmdLine, $rv);
			if($rv!=0) {
				$job->state = $job::STATE_FAIL;
				$this->SaveJob($job);
			}
			else {
				$job->process = (int)kFile::getFileContent($chunk_job_pid_file);
				kFile::unlink($chunk_job_pid_file);
			}
			KalturaLog::log("id:$job->id,keyIdx:$job->keyIdx,rv:$rv,process:$job->process,cmdLine:$cmdLine");
			return true;
		}
		
		/**
		 * This method executed one job fetched from the chunk encoding queue
		 * Please note that changing this methods signature may require adjusting the regex in KillJobsCommand method
		 *
		 * @param $host
		 * @param $port
		 * @param $token
		 * @param $jobIndex
		 * @param string $tmpPromptFolder
		 * @return bool
		 */
		public static function ExecuteJobCommand($host, $port, $token, $jobIndex, $tmpPromptFolder="/tmp",$ffmpegBin="ffmpeg",$ffprobeBin="ffprobe")
		{
			KalturaLog::log("host:$host, port:$port, token:$token, jobIndex:$jobIndex");
			$storeManager = new KChunkedEncodeMemcacheWrap($token);
				// 'flags=1' stands for 'compress stored data'
			$config = array('host'=>$host, 'port'=>$port, 'flags'=>1);
			$storeManager->Setup($config);
			
			if(!isset($jobIndex)) {
				return false;
			}

			$job = $storeManager->FetchJob($jobIndex);
			if($job===false)
				return false;
			
			kBatchUtils::tryLoadKconfConfig();
			
			$job->startTime = time();
			$job->process = getmypid();
			$job->hostname = gethostname();
			$job->state = $job::STATE_RUNNING;
			$storeManager->SaveJob($job);
			
			$outFilename = null;
			if(is_array($job->cmdLine)) {
				$cmdLine = $job->cmdLine[0];
				$outFilenames = isset($job->cmdLine[1]) ? $job->cmdLine[1] : null;;
				$outFilename = is_array($outFilenames) ? $outFilenames[0] : $outFilenames;
				//Added to support use cases where the shared file system is not NFS but rather a remote object storage.
				$sharedChunkPaths = isset($job->cmdLine[2]) ? $job->cmdLine[2] : null;
//				$outFilename = $job->cmdLine[1];
					/*
					 * Switch chunk generation to local disk folder (tmpPromptFolder)
					 * - replace the out file in tha cmdline to '/tmp' flolder
					 * - move all generated files back to the 'normal' tmp path
					 */
				$outFilenameInfo = null;
					// !!!TO DISABLE chunk-to-tmp flow, turn the 'true' to 'false' in condition bellow!!!
				if(isset($tmpPromptFolder) && true) {
					$outFilenameInfo = pathinfo($outFilename);
					$outFilename=$tmpPromptFolder.'/'.$outFilenameInfo['basename'];
					$cmdLine=str_replace($outFilenameInfo['dirname'], $tmpPromptFolder, $cmdLine);
						//Adjust the 'shared' mode outputs to 'chk-to-tmp' flow
					if(is_array($outFilenames)) {
						$storageFilenames = array();
						foreach($outFilenames as $idx=>$tmpName) {
							$storageFilenames[$idx] = $outFilenames[$idx];
							$tmpNameInfo = pathinfo($tmpName);
							$outFilenames[$idx]=$tmpPromptFolder.'/'.$tmpNameInfo['basename'];
						}
					}
				}
			}
			else
				$cmdLine = $job->cmdLine;
				/*
				 * Up to 2 attempts for chunk generation, in order to overcome possible
				 * zeroed frames issue, caused by too large GOPs in the source files 
				 */
			for($i=0;$i<2;$i++) {
				if($i>0){
					$cmdLine = self::fixCmdLineBackOffset($cmdLine);
					KalturaLog::log("2nd attempt, due to empty frames, fixed cmdLine:$cmdLine");
					sleep(3);
				}
				exec($cmdLine,$op,$rv);
				$job->finishTime = time();
				if($rv!=0) {
					$job->state = $job::STATE_FAIL;
					$rvStr = "FAILED - rv($rv),";
					break;
				}
					// No need to check for empties for 1st chunk and audio chunks
				if($job->id==0 || strstr($outFilename,'.vid')===false)
					break;
				if(KFFMpegMediaParser::detectEmptyFrames($ffmpegBin, $ffprobeBin, $outFilename)===false)
					break;
			}

			if($rv==0) {
				if(isset($outFilename) && strstr($outFilename,'.vid')) {
					$stat = new KChunkFramesStat();
						// VP9 / AV1 requires MP4 file container, they don't comply w/MPEGTS
						// X264 / X265 comply w/both MP4 & MPEGTS, but meanwhile we'll continue w/MPEGTS
					if((strstr($cmdLine,"libvpx-vp9")!==false) || (strstr($cmdLine,"libaom-av1")!==false))
						$rv = $stat->getDataMP4($outFilename,$ffprobeBin,$ffmpegBin,$tmpPromptFolder);
					else
						$rv = $stat->getDataMpegts($outFilename,$ffprobeBin,$ffmpegBin,$tmpPromptFolder);
					$job->stat = $stat;
				}

				if(isset($stat) && $stat->isEmpty()){
					$job->state = $job::STATE_FAIL;
					$rvStr = "FAILED - missing chunk stat,";
					$job->msg = "missing chunk stat";
				}
				else {
					$job->state = $job::STATE_SUCCESS;
					$rvStr = "SUCCESS -";
				}
{'fraction_non_alphanumeric': 0.2054820495533457, 'fraction_numerical': 0.006152031012978257, 'mean_word_length': 2.8118374558303887, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 343, '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': '2362095', 'n_tokens_mistral': 19183, 'n_tokens_neox': 15109, 'n_words': 2970}