The biggest issue of using SQL jobs to schedule tasks is the resource consumption due to the inefficienct SQL cache utilization of SQL Server Agent. Service Broker in this case can be employed to resolve this issue. There are 2 options. One is to implement a tiny version of SQL Server Agent by Service Broker, in which we will discuss in the post, and the other way is to use Service Broker as a task conveyer and executor, which I am going to talk in next few posts in this series.
Now let’s Implement Service Broker as a replacement of SQL Server Agent to schedule T-SQL Tasks. The first, we need to create a database with Service Broker enabled, as code snippet showing below
use master
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
go
alter database test set enable_broker
go
Since we are only intent to replace the scheduling part of the system, the task checking, ordering, and loading parts stay the same.
use test
go
create table Tasks
(
TaskID int identity(1,1) primary key,
ExecutionOrder int null,
TaskBody varchar(max) not null,
Status varchar(20) not null default('Pending') -- can be Pending, Running, and Finished
)
go
create index IX_ExecutionOrder on Tasks(ExecutionOrder)
create index IX_Status on Tasks(Status)
go
create table Logs
(
LogId bigint identity(1,1) primary key,
TaskID int not null,
ExecutionOrder int not null,
TaskBody varchar(max),
SPID int not null default(@@SPID),
StartTime datetime not null default(getdate()),
Endtime datetime
)
go
create procedure ExecuteTask(@TaskID int, @ExecutionOrder int, @TaskBody nvarchar(max))
as
begin
declare @LogID bigint
insert into Logs(TaskID, TaskBody, ExecutionOrder)
values(@TaskID, @TaskBody, @ExecutionOrder)
select @LogID = scope_identity()
begin try
exec(@TaskBody )
end try
begin catch
-- your error handling logics
end catch
update Logs set EndTime = getdate() where LogID = @LogID
end
go
create procedure TaskScheduler
(
@WaitingTime int = 30 -- when queue is empty, it waits for 30 second by default before quitting.
)
as
begin
declare @StartTime datetime, @TaskID int, @TaskBody nvarchar(max), @ExecutionOrder int
select @StartTime = getdate()
while(1=1)
begin
begin transaction
update top (1) t
set @TaskID = TaskID,
@TaskBody = TaskBody,
@ExecutionOrder = ExecutionOrder,
Status = 'Running'
from Tasks t with (readpast)
where t.Status = 'Pending'
and not exists(
select *
from Tasks t1 with (nolock)
where t1.ExecutionOrder < t.ExecutionOrder
)
if @@rowcount = 1
begin
exec ExecuteTask @TaskID, @ExecutionOrder, @TaskBody
delete from Tasks where TaskID = @TaskID
select @StartTime = getdate()
end
else
waitfor delay '00:00:00:100' -- wait for 100 milliseconds
commit -- we can commit anyways here
if datediff(second, @StartTime, getdate()) > @WaitingTime
break; -- if there is no task in the queue in last @WaitingTime seconds
end
end
go
create procedure ExecAsync (@SQL varchar(max), @ExecutionOrder int)
as
begin
insert into Tasks(TaskBody, ExecutionOrder) values(@SQL, @ExecutionOrder)
end
go
Now let’s do implementation part of Service Broker
use test
go
create queue QueueScheduler
create service ServiceScheduler on queue QueueScheduler([DEFAULT])
go
go
create procedure ActivationQueueScheduler
as
begin
declare @Handle uniqueidentifier, @Type sysname, @msg nvarchar(max)
waitfor(
Receive top (1)
@Handle = conversation_handle,
@Type = message_type_name
from QueueScheduler
), timeout 5000 -- wait for 5 seconds
if @Handle is null -- no message received
return;
if @Type = 'http://schemas.microsoft.com/SQL/ServiceBroker/DialogTimer'
begin
begin conversation timer (@Handle) timeout = 10;
exec TaskScheduler
end
else
end conversation @Handle;
end
go
alter queue QueueScheduler
with status = on , retention = off ,
activation (
status = on ,
procedure_name = ActivationQueueScheduler,
max_queue_readers = 3,
execute as 'dbo'
)
go
We create Queue and Service associated with the queue, then create Activation procedure going to be associated with the queue. Within the activation procedure, we try to get a message from the queue. If there is no message left in the queue, the activation procedure quit. If there is an message with type http://schemas.microsoft.com/SQL/ServiceBroker/DialogTimer, then we send a timer message to the queue. The timer message sent, in this case, will come back to the queue after 10 seconds. Then it calls procedure TaskScheduler to load the tasks. This keeps the message loop alive.
If message received is not a timer message, such as error message or end of conversation message, we end the conversation. In this case, the Service Broker only acts as a timer rather than a message conveyer. Now let’s create 2 procedures to Activate the scheduler and stop the scheduler.
create procedure StartScheduler
as
begin
declare @handle uniqueidentifier
select @handle = conversation_handle
from sys.conversation_endpoints
where is_initiator = 1
and far_service = 'ServiceScheduler'
and state <> 'CD'
if @@ROWCOUNT = 0
begin
begin dialog conversation @handle
from service ServiceScheduler
to service 'ServiceScheduler'
on contract [DEFAULT]
with encryption = off;
declare @i int = 0, @MaxReaders int = 0
select @MaxReaders = max_readers from sys.service_queues where name = 'QueueScheduler'
while @i< @MaxReaders
begin
begin conversation timer (@Handle) timeout = 1;
select @i = @i + 1
end
end;
end
go
create procedure StopScheduler
as
begin
declare @handle uniqueidentifier
select @handle = conversation_handle
from sys.conversation_endpoints
where is_initiator = 1
and far_service = 'ServiceScheduler'
and state <> 'CD'
if @@ROWCOUNT <> 0
end conversation @handle
end
while starting the scheduler, StartScheduler procedure check if there is any active conversation associated with ServiceScheduler. If yes, it means the scheduler is active, there is no further operation needed. Otherwise the procedure creates a new conversation from ServiceScheduler to itself, check the number of activator of the queue, and then send same number of timer messages to the queue. while stopping the scheduler, procedure StopScheduler simply end the conversation.
Now, let’s start the scheduler, wait for a while and check the active activated tasks.
exec StartScheduler
go
--wait for a minute
go
select * from sys.dm_broker_activated_tasks
go
/*
spid database_id queue_id procedure_name execute_as
----------- ----------- ----------- -------------------------------- -----------
18 13 181575685 [dbo].[ActivationQueueScheduler] 1
26 13 181575685 [dbo].[ActivationQueueScheduler] 1
31 13 181575685 [dbo].[ActivationQueueScheduler] 1
(3 row(s) affected)
*/
There are 3 running sessions launched by Service Broker Queue. Then let’s enqueue some tasks and observate execution Logs
exec ExecAsync 'Waitfor delay ''00:00:10.100''', 100
exec ExecAsync 'Waitfor delay ''00:00:10.100''', 100
exec ExecAsync 'Waitfor delay ''00:00:10.200''', 200
exec ExecAsync 'Waitfor delay ''00:00:10.200''', 200
exec ExecAsync 'Waitfor delay ''00:00:10.200''', 200
exec ExecAsync 'Waitfor delay ''00:00:10.300''', 300
exec ExecAsync 'Waitfor delay ''00:00:10.400''', 400
exec ExecAsync 'Waitfor delay ''00:00:10.400''', 400
exec ExecAsync 'Waitfor delay ''00:00:10.500''', 500
--wait for few seconds and then run
select * from Logs with (nolock)
It behaves exactly same as the result from the experiment from last post of this series. There is no dynamic SQL statement overhead issued by the scheduler. This’s definitely an approach to replace SQL Agent jobs for parallel task scheduling. This solution is perfect in most of the cases, however, there is still a big drawback which is that the scheduler uses multiple sessions constantly scan the Tasks table. The process keeps spinning. That creates the hot spot and overhead of processor to the system.
In my next post of this series, I will discuss how to use Service Broker acts as both task conveyer and scheduler to overcome the issue we have in this experiment.