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

PixelPathProviderSchedulerMainLoop.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.

------------------------------------------------------------------------------]]
	
--
local Rx = require 'Rx'

	-- wf.core
local KSDate = require 'KSDate'
local KSTableUtils = require 'KSTableUtils'
local WFArguments = require 'WFArguments'
local WFFunctionUtils = require 'WFFunctionUtils'

	-- oz_client.managed_catalog
local WFCatAssets = require "WFCatAssets"

	-- pixel provider tasks
local PixelProviderLocalBinaryPathCheckTask = 
		require 'PixelProviderLocalBinaryPathCheckTask'
local PixelProviderSaveDownloadedTask = 
		require 'PixelProviderSaveDownloadedTask'
local PixelProviderChooseRenditionStrategyTask = 
		require 'PixelProviderChooseRenditionStrategyTask'
local PixelProviderCheckFullRenditionDownloadLinkStatusTask = 
		require 'PixelProviderCheckFullRenditionDownloadLinkStatusTask'
local PixelProviderChooseBestPossibleLocalNegativePathTask = 
		require 'PixelProviderChooseBestPossibleLocalNegativePathTask'
local PixelProviderDownloadTask = 
		require 'PixelProviderDownloadTask'
local PixelProviderPixelCacheCheckTask = 
		require 'PixelProviderPixelCacheCheckTask'
local PixelProviderRenderSaveTask = 
		require 'PixelProviderRenderSaveTask'
local PixelProviderDocstoreInfoTask = 
		require 'PixelProviderDocstoreInfoTask'
local PixelProviderFetchXmpDevelopTask = 
		require 'PixelProviderFetchXmpDevelopTask'
local PixelProviderAnalyzeXmpSidecarTask =
		require 'PixelProviderAnalyzeXmpSidecarTask'
local PixelProviderRenderTask = 
		require 'PixelProviderRenderTask'
local PixelProviderRenditionDiskStatusCache =
		require 'PixelProviderRenditionDiskStatusCache'
local PixelProviderSendDataTask = 
		require 'PixelProviderSendDataTask'
local PixelProviderSharedTaskUtils = 
		require 'PixelProviderSharedTaskUtils'
local PixelProviderSharedBlobCache =
		require 'PixelProviderSharedBlobCache'
local PixelPathProviderSchedulerPrivateMethods =
		require 'PixelPathProviderSchedulerPrivateMethods'
local PixelPathProviderSchedulerDiagnosticLog =
		require 'PixelPathProviderSchedulerDiagnosticLog'

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

local KSDate_timeNowForAnimation = assert( KSDate.timeNowForAnimation,
											"KSDate.timeNowForAnimation missing" )

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

local KSTableUtils_derefOpt = assert( KSTableUtils.derefOpt,
										"KSTableUtils.derefOpt does not exist" )

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

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

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

local Rx_Observable_just = assert( Rx.Observable and Rx.Observable.just,
									'Rx.Observable.just missing' )

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

local KSTableUtils_extractKeys = assert( KSTableUtils.extractKeys,
											'KSTableUtils.extractKeys' )

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

local WFArguments_check = assert( WFArguments.check,
										"missing WFArguments.check" )

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

local WFFunctionUtils_reduce = assert( WFFunctionUtils.reduce,
			"WFFunctionUtils.reduce does not exist" )

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

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

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

local function runTaskCleanup( taskCleanupInfo, taskStates )

	local taskName, cleanupFunction = unpack( taskCleanupInfo )

	if type( cleanupFunction ) ~= "function" then

		logger:errorf( "invalid cleanupFunction %s", taskName )

		return

	end

	local ok, maybeErr = pcall( cleanupFunction, taskStates[ taskName ] )

	if not ok then

		logger:errorf( "Error running cleanupFunction %s: %s", taskName,
						maybeErr )

	end

end

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

local function runTaskCleanupOnAll( taskStates, taskCleanupInfos )

	return WFFunctionUtils_map( runTaskCleanup, taskCleanupInfos, taskStates )

end

--==============================================================================
-- add support for allowlisting tasks
--==============================================================================

local function wrapInAllowlistCheck( taskName, f )

	return function( scheduler, ... )

		WFArguments_check( 'PixelPathProviderScheduler' )

		local allowlist = KSTableUtils_derefOpt( scheduler, '_testData', 
													'runLoopTaskAllowList' ) 

		if allowlist and not allowlist[ taskName ] then

			return

		end

		return f( scheduler, ... )

	end

end

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

local function pack( ... )

	return { ... }, select( '#', ... )

end

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

local PixelPathProviderSchedulerDiagnosticLog_recordTimingForDiagnosticLog =
		PixelPathProviderSchedulerDiagnosticLog.recordTimingForDiagnosticLog

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

local function wrapInTiming( threshold, desc, f )

	return function( scheduler, ... )

		WFArguments_check( 'PixelPathProviderScheduler' )

		local start = KSDate_timeNowForAnimation()

		local ret, nRet = pack( f( scheduler, ... ) )

		local duration = KSDate_timeNowForAnimation() - start
		
		if duration > threshold then

			logger:warnf( "%s took %.5f sec (more than threshold of %.5f sec)", 
							desc, duration, threshold )

		end

		PixelPathProviderSchedulerDiagnosticLog_recordTimingForDiagnosticLog( 
			scheduler, desc, duration )

		return unpack( ret, 1, nRet )

	end

end

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

local function wrapProcessRequests( taskName, f )

	local allowlistedFunc = wrapInAllowlistCheck( taskName, f )

	local funcName = string.format( "%s.processRequests", taskName )

	local transactionFunc = 
			PixelPathProviderSchedulerPrivateMethods.wrapSchedulerFunctionInTransaction( 
				funcName, allowlistedFunc ) 

	return wrapInTiming( 1 / 10, funcName, transactionFunc )

end

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

local PixelProviderDocstoreInfoTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderDocstoreInfoTask",
		PixelProviderDocstoreInfoTask.processRequests )

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

local PixelProviderFetchXmpDevelopTask_processRequests = 
	wrapProcessRequests(
		"PixelProviderFetchXmpDevelopTask",
		PixelProviderFetchXmpDevelopTask.processRequests )
		
--------------------------------------------------------------------------------

local PixelProviderAnalyzeXmpSidecarTask_processRequests =
	wrapProcessRequests(
		"PixelProviderAnalyzeXmpSidecarTask",
		PixelProviderAnalyzeXmpSidecarTask.processRequests )

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

local PixelProviderPixelCacheCheckTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderPixelCacheCheckTask",
		PixelProviderPixelCacheCheckTask.processRequests )

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

local PixelProviderLocalBinaryPathCheckTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderLocalBinaryPathCheckTask",
		PixelProviderLocalBinaryPathCheckTask.processRequests )

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

local PixelProviderChooseRenditionStrategyTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderChooseRenditionStrategyTask",
		PixelProviderChooseRenditionStrategyTask.processRequests )

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

local PixelProviderCheckFullRenditionDownloadLinkStatusTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderCheckFullRenditionDownloadLinkStatusTask",
		PixelProviderCheckFullRenditionDownloadLinkStatusTask.processRequests )

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

local PixelProviderChooseBestPossibleLocalNegativePathTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderChooseBestPossibleLocalNegativePathTask",
		PixelProviderChooseBestPossibleLocalNegativePathTask.processRequests )

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

local PixelProviderDownloadTask_processRequests =
	wrapProcessRequests( 
		"PixelProviderDownloadTask",
		PixelProviderDownloadTask.processRequests )
	
--------------------------------------------------------------------------------

local PixelProviderRenderTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderRenderTask",
		PixelProviderRenderTask.processRequests )

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

local PixelProviderRenderSaveTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderRenderSaveTask",
		PixelProviderRenderSaveTask.processRequests )

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

local PixelProviderSaveDownloadedTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderSaveDownloadedTask",
		PixelProviderSaveDownloadedTask.processRequests )

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

local PixelProviderSendDataTask_processRequests = 
	wrapProcessRequests( 
		"PixelProviderSendDataTask",
		PixelProviderSendDataTask.processRequests )

--==============================================================================
-- loopBody
--==============================================================================

local function runTaskProcessRequestsFunc( funcManifest, scheduler  )

	local ok, maybeErr = pcall( funcManifest[ 2 ], 
									scheduler, unpack( funcManifest, 3 ) )

	if not ok then

		local errStr = string.format( "unexpected process request error in %s: %s",
										funcManifest[ 1 ], maybeErr )

		logger:error( errStr )

		return

	end

end

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

local PixelPathProviderSchedulerPrivateMethods_getDeletedRequests =
		PixelPathProviderSchedulerPrivateMethods.getDeletedRequests

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

local function destroyRequestCacheItem( cache, pixStreamId, localAssetId )

	cache:destroyCacheItem( pixStreamId, localAssetId )

end

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

local function purgeRequestCaches( request, perRequestCaches )

	local pixStreamId, localAssetId = unpack( request )

	WFFunctionUtils_map( 
		destroyRequestCacheItem,
			perRequestCaches, pixStreamId, localAssetId )

end

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

local function destroyLocalAssetIdCacheItem( cache, localAssetId )

	cache:destroyCacheItem( localAssetId )

end

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

local function purgeLocalAssetIdCaches( localAssetId, localAssetIdCaches )

	WFFunctionUtils_map( 
		destroyLocalAssetIdCacheItem,
			localAssetIdCaches, localAssetId )

end

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

local PixelPathProviderSchedulerPrivateMethods_getLocalAssetIdsWithNoCurrentRequests =
		PixelPathProviderSchedulerPrivateMethods.getLocalAssetIdsWithNoCurrentRequests

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

local PixelPathProviderSchedulerPrivateMethods_purgeDeletedRequests =
		PixelPathProviderSchedulerPrivateMethods.purgeDeletedRequests

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

local PixelPathProviderSchedulerPrivateMethods_purgeLocalAssetIdsWithNoCurrentRequests =
		PixelPathProviderSchedulerPrivateMethods.purgeLocalAssetIdsWithNoCurrentRequests

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

local function loopBody( taskProcessRequestsPipeline, scheduler, sharedCaches,
							sharedRenderCaches )

	-- run all the task functions once
	WFFunctionUtils_map( runTaskProcessRequestsFunc, 
							taskProcessRequestsPipeline, scheduler )

	-- now do shared cached data cleanup for any requests that are no longer
	-- needed.
	WFFunctionUtils_map( 
		purgeRequestCaches,
			PixelPathProviderSchedulerPrivateMethods_getDeletedRequests( 
				scheduler ),
			{ sharedRenderCaches.activeRenditionInfo,
			  sharedCaches.localBinaryPathCache } )

	PixelPathProviderSchedulerPrivateMethods_purgeDeletedRequests( scheduler )

	WFFunctionUtils_map( 
		purgeLocalAssetIdCaches,
			PixelPathProviderSchedulerPrivateMethods_getLocalAssetIdsWithNoCurrentRequests( 
				scheduler ), 
			{ sharedRenderCaches.currRenderSettingsCache,
			  sharedCaches.baseRenderInfoCache } )

	PixelPathProviderSchedulerPrivateMethods_purgeLocalAssetIdsWithNoCurrentRequests( 
		scheduler )

end

--==============================================================================
-- run loop 
--==============================================================================

local function continueAfterYielding()

	return coroutine.yield() ~= "quit" 

end

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

local function maybeExposeCachesForTests( scheduler, sharedCaches, 
											sharedRenderCaches  )
	
	if type( scheduler._testData ) == "table" then

		scheduler._testData.runLoopCaches = {

			sharedCaches = sharedCaches,

			sharedRenderCaches = sharedRenderCaches

		}

	end

end

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

local function maybeExposeTaskStatesForTests( scheduler, taskStates )
	
	if type( scheduler._testData ) == "table" then

		scheduler._testData.runLoopTaskStates = taskStates

	end

end

--==============================================================================
-- processRequests pipeline
-- 
-- This pipeline runs whenever we add new requests to the scheduler or update
-- the state of a request on the scheduler
--==============================================================================

local function validateTaskProcessRequestsPipelineEntry( 
				sawError, taskProcessRequestsPipelineEntry )

	if sawError then

		return sawError

	end

	local funcDesc, func =
			unpack( taskProcessRequestsPipelineEntry )

	if type( funcDesc ) ~= "string" then

		return "missing function description string"

	end

	if type( func ) ~= "function" then

		return "invalid function specified"

	end

end

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

local function setupTaskProcessRequestsPipeline( sharedCaches,
													sharedRenderCaches,
													taskStates )

	-- taskProcessRequestsPipeline is a list of taskProcessRequestsPipelineEntry
	-- tuples
	-- 
	-- taskProcessRequestsPipelineEntry tuple format is:
	-- 
	-- { functionDescriptionStr,
	-- 		processRequestsFunc,
	-- 			... }
	-- 			
	-- Each processRequestsFunc will be invoked as
	-- 
	-- 	processRequestsFunc( scheduler, ... ) where ... are the original
	-- 	arguments specified in the taskProcessRequestsPipelineEntry.

	local taskProcessRequestsPipeline = {

		{ "PixelProviderDocstoreInfoTask_processRequests",
			PixelProviderDocstoreInfoTask_processRequests, 
				taskStates.PixelProviderDocstoreInfoTask,
				sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.currRenderSettingsCache },
	
		{ "PixelProviderPixelCacheCheckTask_processRequests",
			PixelProviderPixelCacheCheckTask_processRequests, 
				taskStates.PixelProviderPixelCacheCheckTask, 
				sharedCaches.renditionDiskStatusCache,
				sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.currRenderSettingsCache,
				sharedRenderCaches.activeRenditionInfoCache },

		{ "PixelProviderLocalBinaryPathCheckTask_processRequests",
			PixelProviderLocalBinaryPathCheckTask_processRequests, 
				taskStates.PixelProviderLocalBinaryPathCheckTask, 
				sharedCaches.localBinaryPathCache,
				sharedCaches.baseRenderInfoCache },
	
		{ "PixelProviderFetchXmpDevelopTask_processRequests",
			PixelProviderFetchXmpDevelopTask_processRequests, 
				taskStates.PixelProviderFetchXmpDevelopTask,
				sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.currRenderSettingsCache },

		{ "PixelProviderAnalyzeXmpSidecarTask_processRequests",
			PixelProviderAnalyzeXmpSidecarTask_processRequests,
				taskStates.PixelProviderAnalyzeXmpSidecarTask,
				sharedCaches.pendingSaveBlobCache,
				sharedCaches.baseRenderInfoCache },

		{ "PixelProviderChooseRenditionStrategyTask_processRequests",
			PixelProviderChooseRenditionStrategyTask_processRequests,
				taskStates.PixelProviderChooseRenditionStrategyTask, 
				sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedRenderCaches.currRenderSettingsCache,
				sharedRenderCaches.updatedDevelopSettingsAssetIds },

		{ "PixelProviderCheckFullRenditionDownloadLinkStatusTask_processRequests",
			PixelProviderCheckFullRenditionDownloadLinkStatusTask_processRequests,
				taskStates.PixelProviderCheckFullRenditionDownloadLinkStatusTask, 
				sharedCaches.baseRenderInfoCache },

		{ "PixelProviderChooseBestPossibleLocalNegativePathTask_processRequests",
			PixelProviderChooseBestPossibleLocalNegativePathTask_processRequests,
				sharedCaches.baseRenderInfoCache },

	-- process saves before we process downloads or renders. This ensures that
	-- we convert equivalent pending downloads and renders to saves
		{ "PixelProviderRenderSaveTask_processRequests",
			PixelProviderRenderSaveTask_processRequests, 
				taskStates.PixelProviderRenderSaveTask, 
				sharedCaches.pendingSaveBlobCache,
				sharedRenderCaches.activeRenditionInfoCache },

		{ "PixelProviderSaveDownloadedTask_processRequests",
			PixelProviderSaveDownloadedTask_processRequests, 
				taskStates.PixelProviderSaveDownloadedTask, 
				sharedCaches.pendingSaveBlobCache,
				sharedCaches.baseRenderInfoCache },

		{ "PixelProviderSendDataTask_processRequests",
			PixelProviderSendDataTask_processRequests, 
				sharedCaches.renditionDiskStatusCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedCaches.baseRenderInfoCache,
				sharedCaches.localBinaryPathCache },

	-- process renders before downloads, this ensures we convert equivalent pending
	-- downloads to renders
		{ "PixelProviderRenderTask_processRequests",
			PixelProviderRenderTask_processRequests, 
				taskStates.PixelProviderRenderTask, 
				sharedCaches.baseRenderInfoCache,
				sharedCaches.pendingSaveBlobCache,
				sharedRenderCaches.activeRenditionInfoCache },
	
		{ "PixelProviderDownloadTask_processRequests",
			PixelProviderDownloadTask_processRequests, 
				taskStates.PixelProviderDownloadTask,
				sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedCaches.pendingSaveBlobCache },

		{ "PixelProviderSendDataTask_processRequests",
			PixelProviderSendDataTask_processRequests, 
				sharedCaches.renditionDiskStatusCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedCaches.baseRenderInfoCache,
				sharedCaches.localBinaryPathCache },

	}

	local maybeErr = 
			WFFunctionUtils_reduce( nil,
				validateTaskProcessRequestsPipelineEntry, 
					taskProcessRequestsPipeline )

	return not maybeErr and taskProcessRequestsPipeline, maybeErr

end

--==============================================================================
-- run
--
-- The run function runs on its own coroutine "thread", which we'll call the
-- runLoop "thread". We'll call the invoking coroutine "thread" the scheduler
-- "thread".
--
-- The run function primary purpose is run processRequests on all requests with
-- "need to execute task" state.
-- 
-- The tasks are scheduled in order specified in
-- setupTaskProcessRequestsPipeline.
--
-- Each tasks' processRequest method usually pushes one or more requests in a
-- task's worker thread and then changes requests' state to a "wait for task
-- result" state.
-- 
-- When none of the processRequests methods change a request state, the runLoop
-- "thread" will yield control back to the scheduler "thread".
--
-- The tasks' worker thread callbacks will then execute in the scheduler
-- "thread".
--
--==============================================================================

local PixelPathProviderSchedulerPrivateMethods_getStateChangeCounter =
		PixelPathProviderSchedulerPrivateMethods.getStateChangeCounter

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

local function run( scheduler, wfCatalog, pixelProviderSettings )

	-- yield immediately when the runLoop couroutine thread initializes because
	-- we will later resume the runLoop coroutine thread when there is actually
	-- work to do. This allows us to delay the setupTask calls until we are
	-- actually going to do work.
	if not continueAfterYielding() then

		return "done"

	end
	
	----------------------------------------------------------------------------
	-- Set up the sharedRenderCaches. sharedCaches are shared among multiple
	-- tasks. If data is unique to a Task, it should be stored in a taskState.
	-- 
	-- TODO bkruse 2015.11.04 
	--
	-- The sharedCaches are holding too much info for positional parameters. Need
	-- to either use fields or hide the positional arguments behind accessor
	-- functions.
	----------------------------------------------------------------------------

	local wfCatalogProxy = wfCatalog:relayObject()

	local pixelsPath = wfCatalogProxy:fileSpec():pixelsPath()

	local docStore = wfCatalogProxy:docStore() 

	local emptyObservable = Rx.Observable.empty()

	local deletedLocalAssetIdsStream =
			WFCatAssets.allRenderableLocalAssetIdsStream( wfCatalog )
				:flatMap(
					function( allRenderableLocalAssetIdsSetTuple )

						local deletedRenderableAssetIdsSet = 
								allRenderableLocalAssetIdsSetTuple[ 3 ]

						if type( deletedRenderableAssetIdsSet ) ~= "table" or
								not next( deletedRenderableAssetIdsSet ) then

							return emptyObservable

						end
						
						return Rx_Observable_just( 
								KSTableUtils_extractKeys( 
									deletedRenderableAssetIdsSet ) )

					end )

	local sharedCaches = {

		baseRenderInfoCache =
			PixelProviderSharedTaskUtils.makeBaseRenderInfoCache(),

		localBinaryPathCache =
			PixelProviderSharedTaskUtils.makeLocalBinaryPathCache(),
		
		pendingSaveBlobCache =
			PixelProviderSharedBlobCache.makePendingSaveBlobCache( scheduler ),

		renditionDiskStatusCache =
			PixelProviderRenditionDiskStatusCache( 
				pixelsPath, deletedLocalAssetIdsStream )

	}

	local sharedRenderCaches = {

		currRenderSettingsCache =
			PixelProviderSharedTaskUtils.makeCurrRenderSettingsCache(),

		activeRenditionInfoCache =
			PixelProviderSharedTaskUtils.makeActiveRenditionInfoCache(),

		updatedDevelopSettingsAssetIds = {},
											
	}

	maybeExposeCachesForTests( scheduler, sharedCaches, sharedRenderCaches )

	----------------------------------------------------------------------------
	-- Set up any shared objects used by the tasks. 
	----------------------------------------------------------------------------

	local fileSpecObj = wfCatalogProxy:fileSpec()
	
	pixelProviderSettings = pixelProviderSettings or {}
	
	-- On the other hand, tasks sometimes need local state as well. Tasks that 
	-- need local state have setup and cleanup functions.
	--
	-- Setup functions are called before the run loop starts.
	----------------------------------------------------------------------------

	local maxConcurrentDownloadsToSchedule = 20
	local maxConcurrentXmpDevelopFetches = 5
	local maxConcurrentXmpSidecarAnalyzers = 5
	
		-- FIXME [jandrew 2016-06-10]:
		--
		-- The WFRxRateLimiter used in the HTTP stack limits the number of concurrent
		-- HTTP requests to 5. However, it's good to queue up more than the max number of 
		-- concurrent requests so that we can quickly process the next batch of requests
		-- without waiting for them to be scheduled. Hence maxConcurrentDownloadsToSchedule == 20.
		--
		-- However, when I bumped up this limit, the pixel provider eventually
		-- grinded to a halt. We downloaded too much stuff, too quickly, for the save 
		-- render task to handle. The workaround was to increase the number of
		-- maxConcurrentSaves from 2 to 10, though this might have some serious performance
		-- impliciations for spinning hard disks.
		--
		-- The theory for why this slow down happened is that the binary blobs are downloaded
		-- and then get queued up to be written to disk. We just keep adding to that queue faster
		-- that we can write the blobs to disk. Eventually, we have so much queued up that 
		-- the system becomes incredibly slow. Why does it become slow? Not sure yet.
	
	local taskStates = {

		PixelProviderDocstoreInfoTask = 
			PixelProviderDocstoreInfoTask.setupTask( 
				scheduler, docStore, sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.currRenderSettingsCache,
				sharedRenderCaches.updatedDevelopSettingsAssetIds ),
				
		PixelProviderPixelCacheCheckTask =
			PixelProviderPixelCacheCheckTask.setupTask( 
				scheduler, pixelsPath,
				sharedCaches.renditionDiskStatusCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedRenderCaches.currRenderSettingsCache ),

		PixelProviderLocalBinaryPathCheckTask =
			PixelProviderLocalBinaryPathCheckTask.setupTask( docStore, pixelsPath, fileSpecObj ), 
		
		PixelProviderFetchXmpDevelopTask = 
			PixelProviderFetchXmpDevelopTask.setupTask( 
				scheduler, wfCatalogProxy:fileSpec(), maxConcurrentXmpDevelopFetches ),
		
		PixelProviderAnalyzeXmpSidecarTask =
			PixelProviderAnalyzeXmpSidecarTask.setupTask( docStore, maxConcurrentXmpSidecarAnalyzers ),

		PixelProviderDownloadTask = 
			PixelProviderDownloadTask.setupTask( 
				scheduler, wfCatalogProxy, 
				sharedCaches.baseRenderInfoCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedCaches.pendingSaveBlobCache,
				maxConcurrentDownloadsToSchedule ),
			
		PixelProviderRenderTask =
			PixelProviderRenderTask.setupTask( pixelProviderSettings ),

		PixelProviderRenderSaveTask =
			PixelProviderRenderSaveTask.setupTask( 
				scheduler, sharedCaches.renditionDiskStatusCache,
				sharedRenderCaches.activeRenditionInfoCache,
				sharedCaches.pendingSaveBlobCache,
				pixelProviderSettings.maxSaves or 10 ),

		PixelProviderSaveDownloadedTask =
			PixelProviderSaveDownloadedTask.setupTask( 
				scheduler, docStore, fileSpecObj, 
				sharedCaches.localBinaryPathCache,
				sharedCaches.pendingSaveBlobCache,
				pixelProviderSettings.maxSaves or 10 ),
		
		PixelProviderChooseRenditionStrategyTask =
			PixelProviderChooseRenditionStrategyTask.setupTask( pixelsPath ),
		
		PixelProviderCheckFullRenditionDownloadLinkStatusTask =
			PixelProviderCheckFullRenditionDownloadLinkStatusTask.setupTask( 
				wfCatalogProxy:ozCatalogSpec() )

	}

	maybeExposeTaskStatesForTests( scheduler, taskStates )

	------------------------------------------------------------
	-- set up the pipeline
	------------------------------------------------------------

	local taskProcessRequestsPipeline, maybePipelineError = 
			setupTaskProcessRequestsPipeline( 
				sharedCaches, sharedRenderCaches, taskStates )

	if not taskProcessRequestsPipeline then

		logger:errorf( "Error is setting up task processRequests pipeline: %s",
						maybePipelineError )

		return "done"

	end

	------------------------------------------------------------
	-- run the loopBody
	------------------------------------------------------------

	-- outer loop yields control to scheduler thread on yielding 
	-- and breaks when the scheduler thread quits
	while true do

		-- inner loop runs until an error or no requests have a state change.
		-- State changes generally only happen when we schedule a task or
		-- respond to a task callback. Task callbacks do not happen on the
		-- runLoop "thread".
		--
		-- TODO 2015.10.23 bkruse
		--
		-- we may want to limit how long we allow inner loop to run.

		while true do

			local initialProcessRequestsChangeCounter =
					PixelPathProviderSchedulerPrivateMethods.getStateChangeCounter( 
						scheduler ) 

			-- loopBody is defined externally so it's easier to wrap it in pcall
			local ok, maybeErr = pcall( 
									loopBody, 
										taskProcessRequestsPipeline, scheduler,
										sharedCaches, sharedRenderCaches,
										taskStates )

			if not ok then

				logger:error( maybeErr )

				break

			end

			if initialProcessRequestsChangeCounter == 
				PixelPathProviderSchedulerPrivateMethods_getStateChangeCounter(
					scheduler ) then

				break

			end

		end

		scheduler:_maybeSendRequestStats()

		scheduler:_DEBUG_maybeSendLocalAssetIdsStates()

		if not continueAfterYielding() then

			break

		end

	end

	----------------------------------------------------------------------------
	-- runLoop "thread" is done
	----------------------------------------------------------------------------
	----------------------------------------------------------------------------
	-- First, clean up the sharedCaches.
	----------------------------------------------------------------------------
	sharedCaches.pendingSaveBlobCache:destroyAll()

	----------------------------------------------------------------------------
	-- Next, clean up the shared objects.
	----------------------------------------------------------------------------
	wfCatalogProxy:close()
	
	----------------------------------------------------------------------------
	-- Finally, clean up the tasks' local states.
	----------------------------------------------------------------------------
	runTaskCleanupOnAll( taskStates, {
		{ "PixelProviderDocstoreInfoTask", 
			PixelProviderDocstoreInfoTask.cleanup },
		{ "PixelProviderFetchXmpDevelopTask",
			PixelProviderFetchXmpDevelopTask.cleanup },
		{ "PixelProviderAnalyzeXmpSidecarTask",
			PixelProviderAnalyzeXmpSidecarTask.cleanup },
		{ "PixelProviderPixelCacheCheckTask", 
			PixelProviderPixelCacheCheckTask.cleanup },
		{ "PixelProviderRenderTask", 
			PixelProviderRenderTask.cleanup },
		{ "PixelProviderRenderSaveTask", 
			PixelProviderRenderSaveTask.cleanup },
		{ "PixelProviderSaveDownloadedTask", 
			PixelProviderSaveDownloadedTask.cleanup },
		{ "PixelProviderDownloadTask",
			PixelProviderDownloadTask.cleanup },
		{ "PixelProviderCheckFullRenditionDownloadLinkStatusTask",
			PixelProviderCheckFullRenditionDownloadLinkStatusTask.cleanup },
		} )

	return "done"

end

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

local function createRunLoopCoroutineThread( scheduler, wfCatalog, pixelProviderSettings )

	local runLoop = coroutine.create( run )

	-- pass initial arguments to runLoop
	coroutine.resume( runLoop, scheduler, wfCatalog, pixelProviderSettings )

	return runLoop

end

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

local function returnLocalIdIfDeleted( scanState, docValue )

	local localAssetId = docValue._localId

	if not docValue._deleted or not localAssetId then

		return scanState

	end
	
	return scanState, localAssetId

end

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

return {

	createRunLoopCoroutineThread = createRunLoopCoroutineThread,

	-- called by docStore:scannedFilteredMappedChangesFeed
	returnLocalIdIfDeleted = returnLocalIdIfDeleted,

}
