--[[----------------------------------------------------------------------------

PixelProviderRenderTask.lua

--------------------------------------------------------------------------------

ADOBE CONFIDENTIAL
------------------
Copyright 2015 Adobe Systems Incorporated
All Rights Reserved.

NOTICE: All information contained herein is, and remain the property of
Adobe Systems Incorporated and its suppliers, if any. The intellectual and
technical concepts contained herein are proprietary to Adobe Systems
Incorporated and its suppliers and may be covered by U.S. and Foreign
Patents, patents in process, and are protected by trade secret or copyright
law. Dissemination of this information or reproduction of this material is
strictly forbidden unless prior written permission is obtained from
Adobe Systems Incorporated.

------------------------------------------------------------------------------]]

	-- WF.core
local KSData = require 'KSData'
local WFFiniteStateMachine = require 'WFFiniteStateMachine'
local WFFunctionUtils = require 'WFFunctionUtils'
local WFPerformanceReport = require 'WFPerformanceReport'

	-- pixel provider
local PixelProviderSharedBlobCache = require 'PixelProviderSharedBlobCache'
local PixelPathProviderStateMachine = require 'PixelPathProviderStateMachine'
local PixelPathProviderSchedulerPrivateMethods = require "PixelPathProviderSchedulerPrivateMethods"
local PixelPathProviderRequestTypes = require "PixelPathProviderRequestTypes"
local WFCatRenderSize = require 'WFCatRenderSize'
local WFPlatformInfo = require 'WFPlatformInfo'
local NmbAcr = require 'NmbAcr'

	-- Rx
local Rx = require 'Rx'

--------------------------------------------------------------------------------

local WFFunctionUtils_map = assert( WFFunctionUtils.map,
			"WFFunctionUtils.map does not exist" )

--------------------------------------------------------------------------------

local logger = require 'WFLogger'( 'PixelProviderRenderTask' )

local WFCrContext = NmbAcr.WFCrContext
local WFVideoFoundationContext = NmbAcr.WFVideoFoundationContext
local NmbCrPathOverrides = require 'NmbCrPathOverrides'

--==============================================================================
-- setupTask
--==============================================================================

local function setupTask( wfCatalog, pixelProviderSettings )
	
	return { 

		auxParentPath = NmbCrPathOverrides.getAuxParentPath( wfCatalog ),

		maxConcurrentSaves = pixelProviderSettings.maxSaves or 10,			-- specified by client, see LmModelForwarder

		maxConcurrentRenders = pixelProviderSettings.maxRenders or 2,		-- specified by client, see LmModelForwarder
		
		maxConcurrentVideoRenders = pixelProviderSettings.maxVideoRenders or 1,

		activeRenders = {},
			
	}

end

--------------------------------------------------------------------------------

local RenderStates = PixelPathProviderStateMachine.RenderStates

--------------------------------------------------------------------------------

local needsRenderStates =
		{ WFFiniteStateMachine.makeRetryableStates(
				PixelPathProviderStateMachine.RenderRetryCount,
				RenderStates.NeedsRender ) }

--==============================================================================
-- shared methods
--==============================================================================

local function sendTestMessage( request, scheduler, msg )

	local pixStreamId, localAssetId = unpack( request )

	scheduler:_sendTestData( msg, pixStreamId, localAssetId )

end

--------------------------------------------------------------------------------

local function activateRender( activeRenders, renderSlot, renderContext,
								requestRenderSize, digest, localAssetId,
								pixStreamId, requestPriority )

	activeRenders[ renderSlot ] = {

		renderContext = renderContext,

		requestRenderSize = requestRenderSize,

		digest = digest,

		localAssetId = localAssetId, 
		
		pixStreamId = pixStreamId,
		
		requestPriority = requestPriority,

	}

	return activeRenders

end

--------------------------------------------------------------------------------

local function maybeReleaseActiveRender( activeRenders, renderSlot )

	if renderSlot == nil then

		return activeRenders

	end

	local activeRender = activeRenders[ renderSlot ]

	activeRenders[ renderSlot ] = nil

	if type( activeRender ) ~= "table" then

		logger:warnf( "No active render in slot %s", renderSlot )

		return activeRenders

	end

	local localAssetId = activeRender.localAssetId
	
	local requestRenderSize = activeRender.requestRenderSize

	local pixStreamId = activeRender.pixStreamId

	local digest = activeRender.digest

	if activeRender.renderContext then

		activeRender.renderContext:close()

		activeRender.renderContext = nil

	end

	return activeRenders, localAssetId, requestRenderSize, pixStreamId, digest 

end

--------------------------------------------------------------------------------

--==============================================================================
-- handle render result
--==============================================================================

local function validateRenderResult( success, renderBlob )

	if success and not KSData.isData( renderBlob ) then

		return false, string.format( "Bad render blob data of type: %s", 
										type( renderBlob ) )

	end

	return success, renderBlob

end

--------------------------------------------------------------------------------

local PixelProviderSharedBlobCache_getEquivalentRenditionRequests =
		PixelProviderSharedBlobCache.getEquivalentRenditionRequests

--------------------------------------------------------------------------------

local PixelPathProviderSchedulerPrivateMethods_getDeletedRequestsWithState =
	PixelPathProviderSchedulerPrivateMethods.getDeletedRequestsWithState

--------------------------------------------------------------------------------

local PixelPathProviderSchedulerPrivateMethods_wasRequestDeleted =
			PixelPathProviderSchedulerPrivateMethods.wasRequestDeleted

--------------------------------------------------------------------------------

local _orphanedRendersWithPendingCancel = {}

--------------------------------------------------------------------------------

local _unexpectedOrphanedRenders = {}

--------------------------------------------------------------------------------

local function addToOrphanedRenderCountBucket( bucket, pixStreamId,
												requestRenderSize )
	
	bucket[ pixStreamId ] = bucket[ pixStreamId ] or {}

	bucket[ pixStreamId ][ requestRenderSize ] =
		( bucket[ pixStreamId ][ requestRenderSize ] or 0 ) + 1

end

--------------------------------------------------------------------------------

local function handleOrphanedRender( scheduler, pixStreamId, localAssetId,
										requestRenderSize, success, renderBlob )

	local renderRequestWasDeleted =
			PixelPathProviderSchedulerPrivateMethods_wasRequestDeleted( 
				scheduler, pixStreamId, localAssetId )

	local logLevel = renderRequestWasDeleted and "infof" or "warnf"

	local reasonStr =
			renderRequestWasDeleted and 
				string.format( "Prior to cancel, a %s rendition render finished for %s on %s", 
								requestRenderSize, localAssetId, pixStreamId ) or
				string.format( "No active or deleted requests exists for a finished %s rendition render for %s on %s", 
								requestRenderSize, localAssetId, pixStreamId )

	if not success then

		logger:warnf( "ignoring failed render because: %s. Failure info: %s", 
						reasonStr, renderBlob )

		return

	end

	local orphanCountBucket = 
			renderRequestWasDeleted and
				_orphanedRendersWithPendingCancel or
				_unexpectedOrphanedRenders

	addToOrphanedRenderCountBucket( 
		orphanCountBucket, pixStreamId, requestRenderSize )

	logger[ logLevel ]( logger, "destroying render blob because: %s", 
						reasonStr )

	renderBlob:close()

end

--------------------------------------------------------------------------------

local negativeBasedRenderStates =
	{ WFFiniteStateMachine.makeRetryableStates( 
			PixelPathProviderStateMachine.RenderRetryCount, 
					RenderStates.NeedsRender,
					RenderStates.RenderScheduled ) }

--------------------------------------------------------------------------------
-- handleRender
--------------------------------------------------------------------------------

local function handleRenderError( scheduler, requestsSatisfiedByRender,
									localAssetId, requestRenderSize,
									errorMessage )

	-- Changed from errorf to warnf because highly repetative failure rate
	-- with error message as "unknown error" observed in splunk for specific
	-- assets maybe related to TODO below.
	logger:warnf( "Rendition render size %s fail for localAssetId %s: %s",
					requestRenderSize, localAssetId, errorMessage )

	WFFunctionUtils_map( sendTestMessage, 
							requestsSatisfiedByRender,
							scheduler,
							'renderFail' )

	-- TODO bkruse 2017.07.20 
	-- 
	-- it doesn't make sense to retry renders, we should transition to error state

	scheduler:_applyTransitionAllRequests( 
				requestsSatisfiedByRender, 'retry_render' )

	return

end

--------------------------------------------------------------------------------

local function clearDevSettings( request, activeRenditionInfoCache )

	local pixStreamId, localAssetId = unpack( request )

	local activeRenditionInfo = activeRenditionInfoCache:peekCacheItem( 
													pixStreamId, localAssetId ) 

	if type( activeRenditionInfo ) ~= 'table' then

		return

	end

	activeRenditionInfo.devSettings = nil

	activeRenditionInfoCache:pushCacheItem( pixStreamId, localAssetId,
												activeRenditionInfo )

end

--------------------------------------------------------------------------------

local function handleRenderImp( scheduler, success, renderBlob, taskState, 
								renderSlot, activeRenditionInfoCache,
								pendingSaveBlobCache )

	local updatedActiveRenders, localAssetId, requestRenderSize, 
			pixStreamId, digest =
				maybeReleaseActiveRender( taskState.activeRenders, renderSlot )

	taskState.activeRenders = updatedActiveRenders

	if not localAssetId then

		return

	end

	success, renderBlob = validateRenderResult( success, renderBlob )

	local requestsSatisfiedByRender =
		PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
			scheduler, activeRenditionInfoCache, localAssetId,
			requestRenderSize, digest, unpack( negativeBasedRenderStates ) )

	if not next( requestsSatisfiedByRender ) then

		handleOrphanedRender( scheduler, pixStreamId, localAssetId,
								requestRenderSize, success, renderBlob )

		return

	end

	if not success then

		handleRenderError( scheduler, requestsSatisfiedByRender, localAssetId,
							requestRenderSize, renderBlob )

		return

	end

	logger:debugf( "done wth render of size %s for %s", requestRenderSize,
					localAssetId )

	-- clear the dev settings now that we've render they can be big
	WFFunctionUtils_map( clearDevSettings, 
							requestsSatisfiedByRender,
							activeRenditionInfoCache )

	-- retain the blob in a cache so it can later be saved
	pendingSaveBlobCache:pushBlob( localAssetId, renderBlob, 'rendition',
									requestRenderSize, digest )

	-- tell all the requests we've completed a render
	scheduler:_applyTransitionAllRequests( requestsSatisfiedByRender,
											'rendered' )

	WFFunctionUtils_map( sendTestMessage, 
							requestsSatisfiedByRender,
							scheduler,
							'didRender' )

end

--------------------------------------------------------------------------------

handleRenderImp =
	PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInBatchScheduleRequests(
			"PixelProviderRenderTask.handleRenderImp", handleRenderImp, 1 / 6 )

--------------------------------------------------------------------------------

local function handleRender( scheduler, success, renderBlob, taskState, 
								renderSlot, activeRenditionInfoCache,
								pendingSaveBlobCache )

	handleRenderImp( scheduler, success, renderBlob, taskState, renderSlot,
						activeRenditionInfoCache, pendingSaveBlobCache )

end

--==============================================================================
-- processRequests
--==============================================================================

local SaveRenderStates = PixelPathProviderStateMachine.SaveRenderStates

--------------------------------------------------------------------------------

local savingRenderStates =
		{ WFFiniteStateMachine.makeRetryableStates(
				PixelPathProviderStateMachine.SaveRenderRetryCount, 
				SaveRenderStates.SavingRender ) }

--------------------------------------------------------------------------------

local PixelPathProviderSchedulerPrivateMethods_countRequestsWithState =
		PixelPathProviderSchedulerPrivateMethods.countRequestsWithState

--------------------------------------------------------------------------------

local function hasExcessivePendingSaveBacklog( scheduler, taskState )

	local pendingSaveCount = 
			PixelPathProviderSchedulerPrivateMethods_countRequestsWithState( 
				scheduler,  unpack( savingRenderStates ) )

	if pendingSaveCount >= taskState.maxConcurrentSaves then

		scheduler:_sendTestData( 'skipRenderSaveBacklog' )

		return

	end

end

--------------------------------------------------------------------------------
-- maybeJoinRenders
--------------------------------------------------------------------------------

local function maybeJoinPendingRequestsToActiveRender( 
				scheduler, activeRender, activeRenditionInfoCache )

	local localAssetId = activeRender.localAssetId

	local requestRenderSize = activeRender.requestRenderSize

	local digest = activeRender.digest

	-- gather all the pending requests that will also be satisfied by
	-- this active render and join them in waiting

	local equivalentRenderRequests =
			PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
				scheduler, activeRenditionInfoCache, localAssetId,
				requestRenderSize, digest, unpack( needsRenderStates ) )

	scheduler:_applyTransitionAllRequests( equivalentRenderRequests, 'start_render' )

end

--------------------------------------------------------------------------------

local function maybeJoinRenders( scheduler, taskState,
									activeRenditionInfoCache )

	for _, activeRender in pairs( taskState.activeRenders ) do

		maybeJoinPendingRequestsToActiveRender( 
			scheduler, activeRender, activeRenditionInfoCache )

	end

end

--------------------------------------------------------------------------------
-- maybeCancelActiveRender
--------------------------------------------------------------------------------

local function maybeCancelActiveRender( taskState, cancelPredicate, ... )

	local canceledRenderSlot

	local activeRenderSlot, activeRender = next( taskState.activeRenders )

	while activeRenderSlot and not canceledRenderSlot do

		if cancelPredicate( activeRender, ... ) then

			-- disposing the render subscription is how we cancel a render that
			-- is happening on a worker thread
			activeRender.renderSubscription:dispose()

			canceledRenderSlot = activeRenderSlot

		end

		activeRenderSlot, activeRender = 
			next( taskState.activeRenders, activeRenderSlot )

	end

	if canceledRenderSlot then
		
		taskState.activeRenders = 
			maybeReleaseActiveRender( 
				taskState.activeRenders, canceledRenderSlot )

		return true

	end

end

--------------------------------------------------------------------------------
-- maybeCancelOrphanedRenders
--------------------------------------------------------------------------------

local currentlyRenderingStates =
	{ WFFiniteStateMachine.makeRetryableStates( 
			PixelPathProviderStateMachine.RenderRetryCount, 
					RenderStates.RenderScheduled ) }

--------------------------------------------------------------------------------

local function maybeCancelOrphanedRender( activeRender,
											deletedRenderLocalAssetId,
											deletedRenditionInfo ) 

	if deletedRenderLocalAssetId ~= activeRender.localAssetId then

		return false

	end

	if deletedRenditionInfo.renderSize ~= activeRender.requestRenderSize then

		logger:infof( "Won't cancel orphaned render %s sized render of %s because deleted cache data render size is %s",
						activeRender.requestRenderSize,
						deletedRenderLocalAssetId,
						deletedRenditionInfo.renderSize )

		return false

	end

	if deletedRenditionInfo.digest ~= activeRender.digest then

		logger:infof( "Won't cancel orphaned render %s sized render of %s because deleted cache data digest data differs from render",
						activeRender.requestRenderSize,
						deletedRenderLocalAssetId,
						deletedRenditionInfo.renderSize )

		return false

	end

	return true

end

--------------------------------------------------------------------------------

local _canceledRenderCount = 0

--------------------------------------------------------------------------------

local function cancelRenderIfOrphaned( deletedRenderRequest, scheduler,
										taskState, activeRenditionInfoCache )

	local deletedPixStreamId, deletedRenderLocalAssetId = 
			unpack( deletedRenderRequest )

	local deletedRenditionInfo = 
			activeRenditionInfoCache
				:peekCacheItem( deletedPixStreamId, deletedRenderLocalAssetId )

	if not deletedRenditionInfo then

		-- 2018.03.28 bkruse
		-- 
		-- This error is probably a race condition: we are cleaning up data
		-- associated with deleted requests before we have a chance to cancel
		-- async work assoicated with them
		logger:warnf( "Unable to determine if deleted render of %s on %s is orphaned because render info has been deleted",
						deletedRenderLocalAssetId, deletedPixStreamId )

		return

	end

	local equivalentActiveRenderRequests =
			PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
				scheduler, activeRenditionInfoCache,
				deletedRenderLocalAssetId, deletedRenditionInfo.renderSize,
				deletedRenditionInfo.digest, unpack( currentlyRenderingStates ) )

	if #equivalentActiveRenderRequests > 0 then

		logger:infof( "Will not cancel %s sized render of %s because other requests still active",
						deletedRenditionInfo.renderSize, deletedRenderLocalAssetId )

		return

	end

	local didCancel =
		maybeCancelActiveRender( taskState,
			maybeCancelOrphanedRender, 
				deletedRenderLocalAssetId, deletedRenditionInfo )

	if didCancel then

		_canceledRenderCount = _canceledRenderCount + 1

		logger:infof( "Canceled an active render %s sized render request of %s because there are no active requests associated with it",
						deletedRenditionInfo.renderSize, deletedRenderLocalAssetId )

	else

		-- This happens when render finishes before cancel attempt.
		logger:infof( "There was no matching active render for deleted %s sized render request of %s on %s",
						deletedRenditionInfo.renderSize,
						deletedRenderLocalAssetId, deletedPixStreamId )

	end

end

--------------------------------------------------------------------------------

local function maybeCancelOrphanedRenders( scheduler, taskState,
											activeRenditionInfoCache )

	local deletedRenderRequests =
			PixelPathProviderSchedulerPrivateMethods_getDeletedRequestsWithState(
				scheduler,
				unpack( currentlyRenderingStates ) )

	WFFunctionUtils_map(
		cancelRenderIfOrphaned,
			deletedRenderRequests, scheduler, taskState,
			activeRenditionInfoCache )

end

--------------------------------------------------------------------------------
-- get next render request support
--------------------------------------------------------------------------------

local function maybeGetNextRenderRequest( scheduler )

	local request =
			{ PixelPathProviderSchedulerPrivateMethods.getNextRequestWithState( 
				scheduler,  unpack( needsRenderStates ) ) }

	local pixStreamId, localAssetId, _, requestPriority = 
			unpack( request )

	if not pixStreamId then

		-- nothing needs render
		return

	end

	return request, pixStreamId, localAssetId, requestPriority

end

--------------------------------------------------------------------------------

local function maybeGetRenderPathsForRequest( scheduler, baseRenderInfoCache, 
												request, pixStreamId,
												localAssetId )

	local baseRenderInfo = baseRenderInfoCache:peekCacheItem( localAssetId )

	local maybeOriginalPath = baseRenderInfo and baseRenderInfo.originalPath

	local maybeProxyPath = baseRenderInfo and baseRenderInfo.proxyPath

	if not ( maybeOriginalPath or maybeProxyPath ) then

		logger:warnf( "Can't render %s.%s: no suitable negatives",
						pixStreamId, localAssetId )

		scheduler:_sendTestData( 'skipRenderNoPath', pixStreamId, localAssetId )

		-- force it to retrieve again
		scheduler:_applyTransitionToRequest( request, 'new_docstore_render_info' )

		return

	end

	return request, baseRenderInfo, maybeOriginalPath, maybeProxyPath

end

--------------------------------------------------------------------------------

local function maybeGetRenderParamsForRequest( scheduler,
												activeRenditionInfoCache,
												request, pixStreamId,
												localAssetId )

	local activeRenditionInfo =
			activeRenditionInfoCache:peekCacheItem( pixStreamId, localAssetId ) 

	if not activeRenditionInfo then

		logger:warnf( "Can't render %s.%s: no suitable devSettings",
						pixStreamId, localAssetId )

		scheduler:_sendTestData( 'skipRenderNoSettings', pixStreamId, localAssetId )

		-- force it to retrieve again
		scheduler:_applyTransitionToRequest( request, 'new_docstore_render_info' )

		return

	end

	return request, activeRenditionInfo

end

--------------------------------------------------------------------------------

local WFCatRenderSize_canRenderWithProxy = WFCatRenderSize.canRenderWithProxy

--------------------------------------------------------------------------------

local function maybeGetNegativePathForRequest( scheduler, request,
												requestRenderSize,
												maybeOriginalPath,
												maybeProxyPath, pixStreamId,
												localAssetId )

	local negativePath = maybeOriginalPath

	if maybeProxyPath and 
			WFCatRenderSize_canRenderWithProxy( requestRenderSize ) then

		negativePath = maybeProxyPath

	end

	if not negativePath then

		logger:warnf( "Can't render %s.%s: no suitable negative, forcing download",
						pixStreamId, localAssetId )

		scheduler:_sendTestData( 'skipRenderNoSuitablePath', pixStreamId, localAssetId )

		-- force it to run through the loop again
		scheduler:_applyTransitionToRequest( request, 'no_local_negative' )

		return

	end

	return request, negativePath

end

--------------------------------------------------------------------------------

local function maybeGetNumericRenderSize( scheduler, request, pixStreamId,
											localAssetId, requestRenderSize,
											croppedLongSide )

	local numericRenderSize = type( requestRenderSize ) == "number" and  
									requestRenderSize or croppedLongSide

	if type( numericRenderSize ) ~= 'number' then

		logger:warnf( "Can't render %s.%s: no numeric render size, giving up",
						pixStreamId, localAssetId )

		scheduler:_sendTestData( 'skipRenderNoRenderSize', pixStreamId, localAssetId )

		scheduler:_applyTransitionToRequest( request, 'no_render_size' )

		return

	end

	return request, numericRenderSize

end

--------------------------------------------------------------------------------

local function maybeGetRenderableRequest( scheduler, scheduledRequest,
											baseRenderInfoCache,
											activeRenditionInfoCache,
											pixStreamId, localAssetId )

	-- general pattern is: can't render request? Then apply "can't render"
	-- transition and return nil

	local requestWithPath, baseRenderInfo, maybeOriginalPath, maybeProxyPath =
			maybeGetRenderPathsForRequest( scheduler, baseRenderInfoCache, 
												scheduledRequest, pixStreamId,
												localAssetId )

	if not requestWithPath then

		return

	end

	local requestWithPathAndParams, activeRenditionInfo =
			maybeGetRenderParamsForRequest( scheduler,
												activeRenditionInfoCache,
												requestWithPath, pixStreamId,
												localAssetId )

	if not requestWithPathAndParams then

		return

	end

	local requestWithNegPathAndParams, negativePath =
			maybeGetNegativePathForRequest( scheduler, requestWithPathAndParams,
												activeRenditionInfo.renderSize,
												maybeOriginalPath,
												maybeProxyPath, pixStreamId,
												localAssetId )

	if not requestWithNegPathAndParams then

		return

	end

	local requestWithNegPathAndParamsAndRenderSize, numericRenderSize =
			maybeGetNumericRenderSize( 
				scheduler, requestWithNegPathAndParams, pixStreamId,
				localAssetId, activeRenditionInfo.renderSize,
				activeRenditionInfo.croppedLongSide )

	if not requestWithNegPathAndParamsAndRenderSize then

		return

	end

	local normalizedRenderSize = 
			numericRenderSize == activeRenditionInfo.croppedLongSide and
				"full" or numericRenderSize

	return requestWithNegPathAndParamsAndRenderSize, negativePath,
			baseRenderInfo.subtype, numericRenderSize,
			activeRenditionInfo.renderSize, normalizedRenderSize,
			activeRenditionInfo.devSettings,
			activeRenditionInfo.videoDevSettings,
			activeRenditionInfo.videoVersion,
			activeRenditionInfo.userOrientation, 
			activeRenditionInfo.temporal, activeRenditionInfo.digest

end

--------------------------------------------------------------------------------
-- concurrent render and render pre-emption support
--------------------------------------------------------------------------------

local idlePriority =
		PixelPathProviderRequestTypes.getPriority(
			PixelPathProviderRequestTypes.names.idle )

--------------------------------------------------------------------------------

local PixelPathProviderRequestTypes_comparePriorities =
		PixelPathProviderRequestTypes.comparePriorities

--------------------------------------------------------------------------------

local function maybeGetUnusedRenderSlot( isBlackoutActive, requestPriority,
											maxConcurrentImageRenders,
											maxConcurrentVideoRenders,
											activeRenders, isVideo )

	local maybeFirstUnusedSlot

	local usedImageSlotCount = 0
	
	local usedVideoSlotCount = 0
	
	-- SPAM [btran 2021-01-13]
	-- Video exporter can only run one instance at a time. Due to this we must limit
	-- the current amount of concurrent video asset renders is 1 render at a time.
	-- If this limit increases in the future, update `setupTask` to account for the new limit.
		
	if isVideo then
		for renderSlot = 1, maxConcurrentVideoRenders do
			
			local videoSlotKey = string.format( "VideoRenderSlot%s", renderSlot )
			
			if not activeRenders[ videoSlotKey ] then

				maybeFirstUnusedSlot = maybeFirstUnusedSlot or 
										videoSlotKey

			else

				usedVideoSlotCount = usedVideoSlotCount + 1

			end

		end
	else
		for renderSlot = 1, maxConcurrentImageRenders do
			
			local imageSlotKey = string.format( "ImageRenderSlot%s", renderSlot )
			
			if not activeRenders[ imageSlotKey ] then

				maybeFirstUnusedSlot = maybeFirstUnusedSlot or 
										imageSlotKey

			else

				usedImageSlotCount = usedImageSlotCount + 1

			end

		end
	end

	-- allow only 1 render when requesting idle render, otherwise allow
	-- multiple
	-- 
	-- in effect this means idle requests are never processed until
	-- all non-idle requests are complete
	local idleRenderLimit = ( isBlackoutActive or
				PixelPathProviderRequestTypes_comparePriorities( 
					requestPriority, idlePriority ) <= 0 ) and 
					1 
	local maxImageAllowedRenders = idleRenderLimit or maxConcurrentImageRenders
	
	local maxVideoAllowedRenders = idleRenderLimit or maxConcurrentVideoRenders
	
	if isVideo and usedVideoSlotCount >= maxVideoAllowedRenders then
		
		return

	elseif not isVideo and usedImageSlotCount >= maxImageAllowedRenders then

		return

	end

	return maybeFirstUnusedSlot

end

--------------------------------------------------------------------------------

local PixelPathProviderRequestTypes_getName =
		PixelPathProviderRequestTypes.getName

--------------------------------------------------------------------------------

local PixelPathProviderSchedulerPrivateMethods_applyRequestTransitionOnScheduler =
		PixelPathProviderSchedulerPrivateMethods.applyRequestTransitionOnScheduler

--------------------------------------------------------------------------------

local _preemptedRenderCount = 0

--------------------------------------------------------------------------------

local function maybePreemptRender( activeRender, scheduler, 
									activeRenditionInfoCache,
									incomingRenderPriority,
									incomingPixstreamId, incomingLocalAssetId,
									incomingRenderSize )

	local activeRenderRequestPriority = activeRender.requestPriority

	if PixelPathProviderRequestTypes_comparePriorities( 
			incomingRenderPriority, activeRenderRequestPriority ) > 0 then

		_preemptedRenderCount = _preemptedRenderCount + 1

		local activeRenderPixStreamId = activeRender.pixStreamId

		local activeRenderLocalAssetId = activeRender.localAssetId 

		local activeRenderRenderSize = activeRender.requestRenderSize

		local activeRenderDigest = activeRender.digest

		logger:infof( "preempting %s size render of %s by %s at priority '%s' for a %s size render of %s by %s at priority '%s'", 
						activeRenderRenderSize, 
						activeRenderLocalAssetId, 
						activeRenderPixStreamId, 
						PixelPathProviderRequestTypes_getName( activeRenderRequestPriority ),
						incomingRenderSize, incomingLocalAssetId,
						incomingPixstreamId, 
						PixelPathProviderRequestTypes_getName( incomingRenderPriority ) )

		local equivalentActiveRenderRequests =
				PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
					scheduler, activeRenditionInfoCache,
					activeRenderLocalAssetId, activeRenderRenderSize,
					activeRenderDigest, unpack( currentlyRenderingStates ) )

		WFFunctionUtils_map(
			PixelPathProviderSchedulerPrivateMethods_applyRequestTransitionOnScheduler,
				equivalentActiveRenderRequests, scheduler, 
				'reschedule_render' )

		scheduler:_sendTestData( 'renderPreempted' )

		return true
		
	end
		
	return false

end

--------------------------------------------------------------------------------

local function maybePreemptRenders( scheduler, activeRenditionInfoCache,
									taskState, requestPriority, pixstreamId,
									localAssetId, requestRenderSize, isVideo )

	if maybeCancelActiveRender( taskState,
			maybePreemptRender, 
				scheduler, activeRenditionInfoCache, requestPriority,
				pixstreamId, localAssetId, requestRenderSize ) then

		return maybeGetUnusedRenderSlot( 
				scheduler:isRenderBlackoutActive(), requestPriority,
				taskState.maxConcurrentRenders, taskState.maxConcurrentVideoRenders, taskState.activeRenders, isVideo )

	end

end

--------------------------------------------------------------------------------

local function maybeGetEquivalentActiveRenderRequests(
				scheduler, renderableRequest, activeRenditionInfoCache,
				pixStreamId, localAssetId, requestRenderSize, digest )

	local equivalentRenderRequests =
			PixelProviderSharedBlobCache_getEquivalentRenditionRequests( 
				scheduler, activeRenditionInfoCache, localAssetId,
				requestRenderSize, digest, unpack( needsRenderStates ) )

	if #equivalentRenderRequests == 0 then

		logger:errorf( "Bad cache state for render of size %s for %s in %s",
						requestRenderSize, localAssetId, pixStreamId )

		scheduler:_applyTransitionToRequest( 
					renderableRequest, 'cannot_render' )

		return

	end

	return equivalentRenderRequests

end

--------------------------------------------------------------------------------
-- render context
--------------------------------------------------------------------------------

local function getWrappedVideoRenderContext( negativePath )

	local renderContext
	renderContext = WFVideoFoundationContext( negativePath )

	return {

			render = 

				function( _, _, _, videoDevelopSettings, videoVersion, _, temporal, longSide, quality )
				
					logger:infof( "Poster Frame Request from wf: %s", negativePath )
					
					return renderContext:render( longSide, quality, videoDevelopSettings, videoVersion, temporal )
							:map( function( result )

								if result and result.success then
									return result.encodedBlob
								end
								
								if type( result ) == 'table' then
									logger:errorf( "VideoFoundation method %s failed to render %s because: %s",
															result.message or "Unknown", negativePath, result.errorMessage or "Unknown" )
								else
									logger:errorf( "VideoFoundation failed to render %s from unknown error.", negativePath )
								end
								
							end)

				end,

			close =

				function()

					renderContext:close()

				end,

		}

end

--------------------------------------------------------------------------------

local function getWrappedImageOrMockRenderContext( negativePath, taskState )

	local makeWFCrContext = taskState.MockWFCrContext or WFCrContext

	local renderContext = makeWFCrContext( negativePath )

	return {

			render = 

				function( _, devSettings, auxParentPath, _, _, orientation, _, longSide, quality ) 

					return renderContext
							:render( devSettings, auxParentPath, orientation,
										longSide, quality )

				end,

			close =

				function()

					renderContext:close()

				end,

		}

end

--------------------------------------------------------------------------------

local function setUpAndInitRenderContext( negativePath, renderSubtype,
											taskState )

	local isVideo = renderSubtype == 'video' or renderSubtype == 'deleted_video'

	local isMockContext = not not taskState.MockWFCrContext

	if isVideo and WFPlatformInfo.useVideoFoundation and not isMockContext then

		return getWrappedVideoRenderContext( negativePath )

	end

	return getWrappedImageOrMockRenderContext( negativePath, taskState )

end

--------------------------------------------------------------------------------

local function maybeGetRenderContext( scheduler, equivalentRenderRequests,
										taskState, pixStreamId, localAssetId,
										renderSubtype, negativePath )

	local ok, renderContext = pcall( 
			setUpAndInitRenderContext, 
				negativePath, renderSubtype, taskState )

	if not ok then

		if tostring( renderContext ):find( "Failed to create WFCrContextImp object" ) then
			-- ACR is shut down, don't mark the asset with a permanent failure so we try again.
			return
		end

		logger:errorf( "Cannot render %s.%s at path '%s': %s",
						pixStreamId, localAssetId, negativePath, renderContext )

		scheduler:_applyTransitionAllRequests( equivalentRenderRequests,
												'cannot_render' )

		WFFunctionUtils_map( sendTestMessage, 
								equivalentRenderRequests,
								scheduler,
								'renderContextFail' )

		return 

	end

	return renderContext

end

--------------------------------------------------------------------------------
-- render stream support
--------------------------------------------------------------------------------

local function prependWith( val )

	return function( ... )

		return val, ...

	end

end

--------------------------------------------------------------------------------

local function compose( f, g )

	return function( ... )

		return f( g( ... ) )

	end

end

--------------------------------------------------------------------------------

local function convertOnErrorToOnData( stream )

	return stream
			:map( prependWith( true ) )
			:catch( compose( Rx.Observable.just, prependWith( false ) ) )

end

--------------------------------------------------------------------------------
-- stamina support
--------------------------------------------------------------------------------

local WFPerformanceReport_tapObservableOnNext =
		WFPerformanceReport.tapObservableOnNext

--------------------------------------------------------------------------------

local function emitRenderSizeMetric( normalizedRenderSize, success, maybeErr )

	return {

		renderSize = normalizedRenderSize,

		failure = not success and ( maybeErr or "unknown error" ) or nil

	}

end

--------------------------------------------------------------------------------

local function wrapInStaminaBenchmarking( renderStream, normalizedRenderSize )

	local emitRenderSize = compose( emitRenderSizeMetric,
										prependWith( normalizedRenderSize ) )

	return WFPerformanceReport_tapObservableOnNext( "PixelProviderRenderTask",
														renderStream,
														emitRenderSize )

end

--------------------------------------------------------------------------------
-- maybeSetupRenderStream
--------------------------------------------------------------------------------

local function maybeSetupRenderStream( scheduler, equivalentRenderRequests,
										taskState, pixStreamId, localAssetId,
										requestPriority, renderContext,
										renderSlot, negativePath,
										numericRenderSize, requestRenderSize,
										normalizedRenderSize, xmpDevSettings, 
										videoDevSettings, videoVersion, userOrientation, temporal, digest )

	local defaultRenderQuality = 0.8

	logger:debugf( "rendering %s at size %s for %s requests",
					localAssetId, numericRenderSize, 
					#equivalentRenderRequests )

	taskState.activeRenders = 
		activateRender( taskState.activeRenders, renderSlot, renderContext,
						requestRenderSize, digest, localAssetId, pixStreamId,
						requestPriority )

	local renderStreamOk, renderStream =
			pcall( renderContext.render, 
					renderContext, xmpDevSettings, taskState.auxParentPath,
					videoDevSettings, videoVersion,
					userOrientation, temporal, numericRenderSize,
					defaultRenderQuality )

	if not renderStreamOk then

		taskState.activeRenders =
			maybeReleaseActiveRender( 
				taskState.activeRenders, renderSlot )

		logger:errorf( "Cannot setup renderStream %s.%s at path '%s': %s",
						pixStreamId, localAssetId, negativePath, renderContext )

		WFFunctionUtils_map( sendTestMessage, 
								equivalentRenderRequests,
								scheduler,
								'renderMethodFail' )

		scheduler:_applyTransitionAllRequests( equivalentRenderRequests,
												'cannot_render' )

		return 

	end

	renderStream = convertOnErrorToOnData( renderStream )

	renderStream = 
		wrapInStaminaBenchmarking( renderStream, normalizedRenderSize )

	return taskState.activeRenders[ renderSlot ], renderStream

end

--------------------------------------------------------------------------------

local _rendersPerLongSide = {}

--------------------------------------------------------------------------------
-- processRequests
--------------------------------------------------------------------------------

local function processRequests( scheduler, taskState, baseRenderInfoCache,
								pendingSaveBlobCache,
								activeRenditionInfoCache )

	-- first, see if we any requests can join in progress renders
	maybeJoinRenders( scheduler, taskState, activeRenditionInfoCache )
	
	-- second, cancel any active renders that no longer have active requests
	maybeCancelOrphanedRenders( scheduler, taskState, 
								activeRenditionInfoCache )
	
	-- don't bother rendering if we have too many pending saves
	if hasExcessivePendingSaveBacklog( scheduler, taskState ) then

		return

	end

	-- check for needs render request before checking if we have idle render
	-- slot. (Even if all the concurrent renders are active, the needs render
	-- request may need to preempt a render.)
	local scheduledRequest, pixStreamId, localAssetId, requestPriority =
			maybeGetNextRenderRequest( scheduler )

	if not scheduledRequest then

		return

	end

	-- make sure the request can actually be rendered
	local renderableRequest, negativePath, renderSubtype, numericRenderSize,
			requestRenderSize, normalizedRenderSize, xmpDevSettings,
			videoDevSettings, videoVersion, userOrientation, temporal, digest = 
				maybeGetRenderableRequest( 
					scheduler, scheduledRequest, baseRenderInfoCache,
					activeRenditionInfoCache, pixStreamId, localAssetId )
					
	local isVideo = renderSubtype == "video"

	if not renderableRequest then

		return

	end

	-- make sure we are allowed to render
	local unusedRenderSlot = 
				maybeGetUnusedRenderSlot( 
					scheduler:isRenderBlackoutActive(), requestPriority,
					taskState.maxConcurrentRenders, taskState.maxConcurrentVideoRenders, taskState.activeRenders, isVideo ) or
				maybePreemptRenders( 
					scheduler, activeRenditionInfoCache, taskState,
					requestPriority, pixStreamId, localAssetId, numericRenderSize, isVideo )

	if not unusedRenderSlot then

		return

	end

	-- find all the requests that will be satisfied by the render we are about
	-- to start
	local equivalentRenderRequests =
			maybeGetEquivalentActiveRenderRequests(
				scheduler, renderableRequest, activeRenditionInfoCache,
				pixStreamId, localAssetId, requestRenderSize, digest )

	if not equivalentRenderRequests then

		return

	end

	-- set up the render context, a native object that will render a photo or
	-- video
	local renderContext = 
			maybeGetRenderContext( scheduler, equivalentRenderRequests,
										taskState, pixStreamId, localAssetId,
										renderSubtype, negativePath )

	if not renderContext then

		return

	end

	-- avoid race conditions: transition all requests before we setup to the
	-- renderStream in case the renderStream does the render on construction
	-- instead of on subscribe
	scheduler:_applyTransitionAllRequests( equivalentRenderRequests, 
												'start_render' )

	-- set up the render stream. In practice, this sets up the actual async context
	-- for the render, but the worker thread won't fire until we subscribe below
	local activeRender, renderStream = 
			maybeSetupRenderStream( scheduler, equivalentRenderRequests,
									taskState, pixStreamId, localAssetId,
									requestPriority, renderContext,
									unusedRenderSlot, negativePath,
									numericRenderSize, requestRenderSize,
									normalizedRenderSize, xmpDevSettings,
									videoDevSettings, videoVersion, userOrientation, temporal, digest )

	if not activeRender then

		return

	end

	-- subscribe to the renderStream. Subscribing is what actually causes the
	-- render to begin on a worker thread.
	activeRender.renderSubscription =
			renderStream
				:subscribe(
					function( success, renderBlob )

						if success then

							_rendersPerLongSide[ normalizedRenderSize ] =
								( _rendersPerLongSide[ normalizedRenderSize ] or 0 ) + 1

						end

						local ok, maybeError =
							pcall( handleRender, 
									scheduler, success, renderBlob, taskState,
									unusedRenderSlot, activeRenditionInfoCache,
									pendingSaveBlobCache )

						if not ok then

							logger:errorf( "error in handleRender: %s", maybeError )

						end

					end,
					function( errStr )

						-- this should never happen, the convertOnErrorToOnData 
						-- call above should prevent this
						logger:errorf( "Unexpected uncaught error: %s", errStr )

					end )

				
end

--==============================================================================
-- cleanup
--==============================================================================

local function cleanup( taskState )

	local activeRenderSlot 

	repeat

		activeRenderSlot = next( taskState.activeRenders ) 

		if activeRenderSlot then

			taskState.activeRenders = 
				maybeReleaseActiveRender( 
					taskState.activeRenders, activeRenderSlot )

		end

	until not activeRenderSlot

	taskState.activeRenders = {}

end

--==============================================================================
-- getRenderStats
--==============================================================================

local function getRenderStats()

	return _rendersPerLongSide, _preemptedRenderCount, 
			_canceledRenderCount, _orphanedRendersWithPendingCancel,
			_unexpectedOrphanedRenders

end

--==============================================================================

return {

	setupTask = setupTask,

	processRequests = processRequests,

	cleanup = cleanup,

	getRenderStats = getRenderStats,

	privateUnitTestExports = {

		maybeGetUnusedRenderSlot = maybeGetUnusedRenderSlot,

	},

}
