Parallel Task Scheduling (6) – Scheduled By Caller

use master
--create test environment
if DB_ID('test') is not null
begin
	alter database test set single_user with rollback immediate
	drop  database test
end
go
create database test
--enable the service broker
alter database test set enable_broker
go
use test
go
--create ExecutionLog table
create table ExecutionLog
(
	LogID int identity(1,1) primary key,
	CallerID int,--Who sent the message
	SchedulerHandle uniqueidentifier, -- from which handle
	SPID int, -- who executed the task
	SQL nvarchar(max), -- task body
	--Start time, when the task is not executed, 
	--this time is the time when caller sends the task.
	--When the task is executed, the StartTime will be the time task started.
	StartTime datetime2(7) default(sysdatetime()), 
	EndTime datetime2(7) --Task completion time
)
go
-- This is for monitoring and cleanup purpose
create index SPID on ExecutionLog(SPID)
	include (CallerID, StartTime, EndTime) 
where (EndTime IS NULL)
go

go
-- this is view we are going to use later to monitor the execution of task series.
create view ExecutionStatus
as
	select c.LogID, b.spid SPID, c.CallerID, c.StartTime, c.EndTime, c.SQL
	from sys.service_queues a
		inner loop join  sys.dm_broker_activated_tasks b on a.object_id = b.queue_id and b.database_id = DB_ID()
		left loop join ExecutionLog c with (nolock, index = SPID) on  c.EndTime is null and b.spid = c.SPID 

go
--select * from ExecutionStatus
go
--Create message type
--When a task sent from Initiator to Taget, the task is tagged with AsyncSent
create message type AsyncSent validation = none

--When a task sent from Taget to Initiator, the task is tagged with AsyncDone
create message type AsyncDone validation = none
go
-- here is the contract as explained above
create contract AsyncContract
(
	AsyncSent sent by initiator,
	AsyncDone sent by target
)
go
--create essential components for initiator
create queue QueueAsyncInitiator
create service SerivceAsyncInitiator
on queue QueueAsyncInitiator(AsyncContract)
go
--create essential components for Target
create queue QueueAsyncTarget
create service SerivceAsyncTarget
on queue QueueAsyncTarget(AsyncContract)
go
----Activation at target, it's also a task executor
create procedure ActivationAsyncTarget
as
begin
	declare @Handle uniqueidentifier, @Message xml, @MessageType nvarchar(256) 
	declare @SQL nvarchar(max), @LogID int
	while(1=1)
	begin
		select @Handle = null
		waitfor(
			receive top(1)
				@Handle = conversation_handle,
				@Message = cast( message_body as xml),
				@MessageType = message_type_name
			from QueueAsyncTarget
		), timeout 30000 
		-- wait for 30 seconds. Because service broker will start threads gragually based on some algorithm
		-- until value of Max Queue Reader. We want a launched thread stay reasonable longer for receiving
		-- next message.
		
		if @Handle is null 
			break;
		if @MessageType in ('http://schemas.microsoft.com/SQL/ServiceBroker/EndDialog', 'http://schemas.microsoft.com/SQL/ServiceBroker/Error')
		begin 
			-- handling EndDialog and Error in the service broker
			end conversation @Handle;
			break;
		end		
		if @MessageType = 'AsyncSent' -- received MSG is a task!
		begin
			select	@LogID = @Message.value('(/Command/@LogID)[1]', 'int'), 
					@SQL = @Message.value('(/Command/@SQL)[1]', 'nvarchar(max)')
			--perform cleanup if there're any terminated processes
			update a
				set EndTime = GETDATE()
			from ExecutionLog a with(index = SPID)
			where SPID = @@SPID
				and EndTime is null
			-- Set task startup time
			update ExecutionLog 
				set StartTime = GETDATE(),
					SPID = @@SPID
			where LogID = @LogID
			
			begin try
				-- do the work. 
				-- This piece of code should not fail. 
				-- You may consider using CLR with different connection
				-- so that any disconnection will be caught by catch block below
				exec(@SQL)
			end try
			begin catch
				--You Error handling code
			end catch
			-- done, log the time
			update ExecutionLog 
				set EndTime = GETDATE()
			where LogID = @LogID
			-- send message back
			select @Message = (select @LogID as [@LogID] for xml path('Command'))
			;send on conversation @handle message type [AsyncDone](@Message)
		end
	end
end
GO

go
--attach Activation Procedure to target queue
alter queue QueueAsyncTarget
	with status = on, 
	activation(
				status = on, 
				procedure_name = ActivationAsyncTarget, 
				-- here I am using a unrealistic value but definitely 
				-- we need bigger value here since all Schedulers 
				-- will share the same set of activation threads
				max_queue_readers = 100, 
				execute as self -- your own security model
				)
go
create procedure GetScheduler @SchedulerHandle uniqueidentifier output
as
begin
	-- begin a conversation
	begin dialog conversation @SchedulerHandle
	from service SerivceAsyncInitiator
	to service 'SerivceAsyncTarget'
	on contract [AsyncContract]
	with encryption = off
end
go
create procedure ReleaseScheduler @SchedulerHandle uniqueidentifier
as
begin 
	-- release the conversation. 
	-- I used try-catch block here since while this procedure is called,
	-- the conversation might be released by DBA manually :) 
	-- Evil DBAs, but your are important :)
	begin try
		end conversation @SchedulerHandle;
	end try
	begin catch
	end catch
end
go
---Send message to service broker
create procedure ExecAsync @SchedulerHandle uniqueidentifier, @SQL nvarchar(max)
as
begin
	declare @Message xml, @LogID int
	insert into ExecutionLog(CallerID, SQL, SchedulerHandle)
		values(@@SPID, @SQL, @SchedulerHandle)
	select @LogID = scope_identity()	
	select @Message = (select @LogID as [@LogID], @SQL as [@SQL] for xml path('Command')) -- @Order is just for information purpose	
	;send on conversation @SchedulerHandle message type [AsyncSent](@Message);		
end
go
-- Synchronization
create procedure Synchronization @SchedulerHandle uniqueidentifier
as
begin
	declare @MessageType sysname
	waitfor(
			receive top(1)
				@MessageType = message_type_name
			from QueueAsyncInitiator
			where conversation_handle = @SchedulerHandle
		), timeout -1
	if @MessageType != 'AsyncDone'
		raiserror('Unexpected message %s received! The conversation should be ended', 16,1, @MessageType)
end
go

--test
declare @SchedulerHandle uniqueidentifier
declare @SQL nvarchar(max) 
-- get a Scheduler handle
exec GetScheduler @SchedulerHandle output
begin try
	select @SQL = 'waitfor delay ''00:00:10.100'''
	-- Run 4 Tasks
	exec ExecAsync @SchedulerHandle, @SQL
	exec ExecAsync @SchedulerHandle, @SQL
	exec ExecAsync @SchedulerHandle, @SQL
	exec ExecAsync @SchedulerHandle, @SQL
	
	-- Call Synchronization 4 times since we queued 4 tasks
	-- Extra execution of this procedure will lead caller wait for ever
	exec Synchronization @SchedulerHandle
	exec Synchronization @SchedulerHandle
	exec Synchronization @SchedulerHandle
	exec Synchronization @SchedulerHandle
	
	
	select @SQL = 'waitfor delay ''00:00:09.200'''	
	exec ExecAsync @SchedulerHandle, @SQL
	exec ExecAsync @SchedulerHandle, @SQL
	
	exec Synchronization @SchedulerHandle
	exec Synchronization @SchedulerHandle
	
	select @SQL = 'waitfor delay ''00:00:08.300'''	
	exec ExecAsync @SchedulerHandle, @SQL
	
	exec Synchronization @SchedulerHandle
	
	
	select @SQL = 'waitfor delay ''00:00:07.400'''	
	exec ExecAsync @SchedulerHandle, @SQL
	exec ExecAsync @SchedulerHandle, @SQL
	
	exec Synchronization @SchedulerHandle
	exec Synchronization @SchedulerHandle
	
	select @SQL = 'waitfor delay ''00:00:06.500'''	
	exec ExecAsync @SchedulerHandle, @SQL
	
	exec Synchronization @SchedulerHandle
end try
begin catch
	-- your error handling code
end catch
--Release scheduler
exec ReleaseScheduler @SchedulerHandle