I\u2019ve been doing a bit of work on Service Broker while building a series of new Pluralsight courses and one of the things I recently found a good use for Service Broker on a client system with a 8TB VLDB was for implementing parallel maintenance tasks using multiple threads.\u00a0 The idea for this implementation came out of the fact that their maintenance window continues to become shorter and shorter the larger their workload increases, while the size of their database continues to get larger and larger.\u00a0 One of the things I noticed while doing a monthly health check of their server was that during the maintenance window, their hardware is no where close to being utilized and there are plenty of resources available due to the serial execution of their maintenance tasks.\u00a0 This specific client uses Ola Hallengren\u2019s maintenance procedures, which we recommend to all of our clients, so I started to think about ways I could make things run in parallel to get better utilization of the hardware during the maintenance window and hopefully reduce the time required to perform the day to day maintenance in their database.<\/p>\n
I wish I could say the first thing I thought of was Service Broker, but it wasn\u2019t I toyed around with the idea of a custom C# app to run Ola\u2019s scripts and log the output commands to the CommandLog table in master, then use multiple threads to run each of the individual tasks, something similar using Powershell instead of a compiled app, and even multiple agent jobs to break things up using different parameters for IndexOptimize. It turns out that these are all over engineering a very basic problem that Service Broker is highly suited towards solving and it only took a little bit of trial and error to create a very robust solution that runs index and statistics maintenance tasks in parallel in SQL Server.<\/p>\n
I chose to implement the Service Broker components for this configuration in the MSDB database for a couple of reasons.\u00a0 First, it is already enabled for Service Broker, so I don\u2019t have to worry about that, second, it is also already TRUSTWORTHY ON which is required by Agent, and third it is owned by SA, so an activation stored procedure marked with EXECUTE AS OWNER, runs as a full sysadmin in the server allowing it to run any maintenance command generated by Ola\u2019s scripts.\u00a0 We need to configure some basic objects for Service Broker in the database; a message type, a contract, a queue and a service.<\/p>\n
\r\nUSE msdb\r\nGO\r\n-- Create the message types\r\nCREATE MESSAGE TYPE\r\n[OlaHallengrenMaintenanceTaskMessage]\r\nVALIDATION = WELL_FORMED_XML;\r\nGO\r\n\r\n-- Create the contract\r\nCREATE CONTRACT [OlaHallengrenMaintenanceTaskContract]\r\n([OlaHallengrenMaintenanceTaskMessage]\r\nSENT BY INITIATOR);\r\nGO\r\n\r\n-- Create the target queue and service\r\nCREATE QUEUE OlaHallengrenMaintenanceTaskQueue;\r\nGO\r\n\r\nCREATE SERVICE\r\n[OlaHallengrenMaintenanceTaskService]\r\nON QUEUE OlaHallengrenMaintenanceTaskQueue\r\n([OlaHallengrenMaintenanceTaskContract]);\r\nGO\r\n<\/pre>\nBuilding an Activation Procedure to Automate Processing<\/h1>\n
With that setup, now we need to build an activation stored procedure to process the messages from the queue and perform the operations generated by Ola\u2019s scripts.\u00a0 At first I thought that this would be pretty straightforward, but it actually required a fair amount of additional logic to prevent poison messages and queue deactivation from occurring when multiple online index rebuilds were attempted on the same table.\u00a0 When this occurs the engine raises Error 1912:<\/p>\n
Could not proceed with index DDL operation on %S_MSG ‘%.*ls’ because it conflicts with another concurrent operation that is already in progress on the object. The concurrent operation could be an online index operation on the same object or another concurrent operation that moves index pages like DBCC SHRINKFILE.<\/p><\/blockquote>\n
which causes the XACT_STATE() of the operation to become \u20131, which only allows the transaction processing the command to be rolled back, leading to a poison message within Service Broker.\u00a0 So to mitigate against that I had to add in a check against sys.dm_tran_locks for any ALTER_INDEX command to identify if a conflicting lock is currently being held and if it is then we just requeue the message and increment the retry count held within the message so that we can incrementally back-off attempting to run the command each time it encounters a conflicting lock using a WAITFOR.\u00a0 Since Ola\u2019s scripts log the command type, database name, object name, and command to be executed to the CommandLog table in the master database, all we need to queue in our message is the ID of the command to be processed by Service Broker, and we can get the rest of the pertinent information from the CommandLog table directly.\u00a0 As a part of each tasks execution in our activation stored procedure we will also update the StartTime and EndTime of the operation for historical trending of execution durations.<\/p>\n
\r\nIF OBJECT_ID('OlaHallengrenMaintenanceTaskQueue_ActivationProcedure') IS NOT NULL\r\nBEGIN\r\n\tDROP PROCEDURE OlaHallengrenMaintenanceTaskQueue_ActivationProcedure;\r\nEND\r\nGO\r\n\r\nCREATE PROCEDURE OlaHallengrenMaintenanceTaskQueue_ActivationProcedure\r\nAS\r\n\r\n DECLARE @conversation_handle UNIQUEIDENTIFIER;\r\n DECLARE @message_body XML;\r\n DECLARE @message_type_name SYSNAME;\r\n DECLARE @Command NVARCHAR(MAX);\r\n DECLARE @ID INT\r\n DECLARE @DBName SYSNAME;\r\n DECLARE @ObjectName SYSNAME;\r\n DECLARE @CommandType NVARCHAR(60);\r\n DECLARE @Retry INT;\r\n DECLARE @FQN NVARCHAR(400);\r\n\r\n WHILE (1=1)\r\n BEGIN\r\n\r\n BEGIN TRANSACTION;\r\n\r\n WAITFOR\r\n ( RECEIVE TOP(1)\r\n @conversation_handle = conversation_handle,\r\n @message_body = message_body,\r\n @message_type_name = message_type_name\r\n FROM OlaHallengrenMaintenanceTaskQueue\r\n ), TIMEOUT 5000;\r\n\r\n IF (@@ROWCOUNT = 0)\r\n BEGIN\r\n ROLLBACK TRANSACTION;\r\n BREAK;\r\n END\r\n\r\n IF @message_type_name = N'OlaHallengrenMaintenanceTaskMessage'\r\n BEGIN\r\n\t\tSELECT @ID = @message_body.value('(CommandLogID)[1]', 'int'),\r\n\t\t\t\t@Retry = ISNULL(@message_body.value('(CommandLogID\/@retry)[1]', 'int'), 0);\r\n\r\n\t \r\n\t\tSELECT @Command = Command, \r\n\t\t\t\t@ObjectName = ObjectName, \r\n\t\t\t\t@DBName = DatabaseName,\r\n\t\t\t\t@FQN = QUOTENAME(DatabaseName) + '.' + QUOTENAME(SchemaName) + '.' + QUOTENAME(ObjectName),\r\n\t\t\t\t@CommandType = CommandType\r\n\t\tFROM master.dbo.CommandLog\r\n\t\tWHERE ID = @ID;\r\n\r\n\t\t-- \tCheck for Index rebuilds if one is already running and requeue the request after waiting.\r\n\t\tIF @CommandType = 'ALTER_INDEX'\r\n\t\tBEGIN\r\n\r\n\t\t\t-- Check if we have an incompatible lock that would lead to a failed execution \r\n\t\t\tIF EXISTS (SELECT 1\r\n\t\t\t\t\t\tFROM sys.dm_tran_locks AS tl\r\n\t\t\t\t\t\tWHERE (request_mode = 'SCH-M' OR \r\n\t\t\t\t\t\t\t-- Concurrent maintenance task doing UpdateStats?\r\n\t\t\t\t\t\t\t(request_mode = 'LCK_M_SCH_S' AND EXISTS (SELECT 1 FROM sys.dm_exec_sessions AS s WHERE is_user_process = 1 AND tl.request_session_id = s.session_id)))\r\n\t\t\t\t\t\t AND resource_associated_entity_id = OBJECT_ID(@FQN)\r\n\t\t\t\t\t\t AND resource_database_id = DB_ID(@DBName) )\r\n\t\t\tBEGIN\r\n\t\t\t\t-- Wait for 5 seconds times the number of retrys to do an incremental backoff\r\n\t\t\t\t-- This will eventually cause all queue readers to die off and serial execution of tasks\r\n\t\t\t\tDECLARE @Delay NVARCHAR(8) = CAST(DATEADD(ss, @Retry*5, CAST('00:00:00'AS TIME)) AS VARCHAR)\r\n\t\t\t\tWAITFOR DELAY @Delay\r\n\r\n\t\t\t\t-- Increment retry count in the message\r\n\t\t\t\tSELECT @message_body = N'<CommandLogID retry="'+CAST(@Retry+1 AS NVARCHAR)+'">'+CAST(@id AS NVARCHAR)+N'<\/CommandLogID>';\r\n\r\n\r\n\t\t\t\t-- Send the message back to the queue for later processing\r\n\t\t\t\t;SEND ON CONVERSATION @conversation_handle\r\n\t\t\t\t\t\tMESSAGE TYPE [OlaHallengrenMaintenanceTaskMessage]\r\n\t\t\t\t\t\t(@message_body);\r\n\r\n\t\t\t\tGOTO SkipThisRun\r\n\t\t\tEND\r\n\t\tEND\r\n\r\n\t\tUPDATE master.dbo.CommandLog\r\n\t\tSET StartTime = GETDATE()\r\n\t\tWHERE ID = @ID;\r\n\r\n\t\tBEGIN TRY \r\n\t\t\tEXECUTE(@Command);\r\n\r\n\t\t\tUPDATE master.dbo.CommandLog\r\n\t\t\tSET EndTime = GETDATE()\r\n\t\t\tWHERE ID = @ID;\r\n\r\n\t\tEND TRY\r\n\t\tBEGIN CATCH\r\n\r\n\t\t\tUPDATE master.dbo.CommandLog\r\n\t\t\tSET EndTime = GETDATE(),\r\n\t\t\t\tErrorMessage = ERROR_MESSAGE(),\r\n\t\t\t\tErrorNumber = ERROR_NUMBER()\r\n\t\t\tWHERE ID = @ID;\r\n\r\n\t\tEND CATCH\r\n\r\n END CONVERSATION @conversation_handle;\r\n END\r\n\r\n\t-- If end dialog message, end the dialog\r\n ELSE IF @message_type_name = N'http:\/\/schemas.microsoft.com\/SQL\/ServiceBroker\/EndDialog'\r\n BEGIN\r\n END CONVERSATION @conversation_handle;\r\n END\r\n\r\n\t-- If error message, log and end conversation\r\n ELSE IF @message_type_name = N'http:\/\/schemas.microsoft.com\/SQL\/ServiceBroker\/Error'\r\n BEGIN\r\n\t\tDECLARE @error INT;\r\n\t\tDECLARE @description NVARCHAR(4000);\r\n\r\n\t\t-- Pull the error code and description from the doc\r\n\t\tWITH XMLNAMESPACES ('http:\/\/schemas.microsoft.com\/SQL\/ServiceBroker\/Error' AS ssb)\r\n\t\tSELECT\r\n\t\t\t@error = @message_body.value('(\/\/ssb:Error\/ssb:Code)[1]', 'INT'),\r\n\t\t\t@description = @message_body.value('(\/\/ssb:Error\/ssb:Description)[1]', 'NVARCHAR(4000)');\r\n\t\t\r\n\t\tRAISERROR(N'Received error Code:%i Description:"%s"', 16, 1, @error, @description) WITH LOG;\r\n\r\n\t\t-- Now that we handled the error logging cleanup\r\n\t\tEND CONVERSATION @conversation_handle;\r\n\tEND\r\n \r\n SkipThisRun: \r\n COMMIT TRANSACTION;\r\n\r\n END\r\nGO\r\n\r\n<\/pre>\nEnabling Parallel Queue Activation<\/h1>\n
The last thing we need to do is bind our stored procedure to the queue for automated activation and to set the MAX_QUEUE_READERS to the number of concurrent parallel tasks we want to have executing on the system at any given point in time.\u00a0 I spent some time testing different values for MAX_QUEUE_READERS and found that 5 was generally sufficient to provide significant reductions in runtimes without creating contention between the tasks being executed.<\/p>\n
\r\n-- Alter the target queue to specify internal activation\r\nALTER QUEUE OlaHallengrenMaintenanceTaskQueue\r\nWITH ACTIVATION\r\n( STATUS = ON,\r\nPROCEDURE_NAME = OlaHallengrenMaintenanceTaskQueue_ActivationProcedure,\r\nMAX_QUEUE_READERS = 5,\r\nEXECUTE AS SELF\r\n);\r\nGO\r\n<\/pre>\nNow all we have to do is setup a job to feed the tasks into the queue as messages and Service Broker will automatically begin processing the tasks in parallel up to the MAX_QUEUE_READERS configured value.\u00a0 To do this, I just used Ola\u2019s scripts to log the commands to the CommandLog table but not execute them and then using a cursor and ROW_NUMBER() definition to split the work up across different tables instead of queuing the commands in the order generated, send each ID as a message to Service Broker to process.<\/p>\n
\r\nDECLARE @MaxID INT\r\nSELECT @MaxID = MAX(ID)\r\nFROM master.dbo.CommandLog;\r\n\r\nSELECT @MaxID = ISNULL(@MaxID, 0)\r\n\r\n-- Load new tasks into the Command Log\r\nEXECUTE master.dbo.IndexOptimize\r\n@Databases = 'AdventureWorks2014',\r\n@FragmentationLow = 'INDEX_REBUILD_OFFLINE',\r\n@FragmentationMedium = 'INDEX_REORGANIZE,INDEX_REBUILD_ONLINE,INDEX_REBUILD_OFFLINE',\r\n@FragmentationHigh = 'INDEX_REBUILD_ONLINE,INDEX_REBUILD_OFFLINE',\r\n@UpdateStatistics = 'ALL',\r\n@FragmentationLevel1 = 5,\r\n@FragmentationLevel2 = 30,\r\n@LogToTable = 'Y',\r\n@Execute = 'N';\r\n\r\nDECLARE @NewMaxID INT\r\nSELECT @NewMaxID = MAX(ID)\r\nFROM master.dbo.CommandLog;\r\n\r\nUSE\u00a0\u00a0\u00a0 msdb;\r\n\r\nDECLARE @id int;\r\n\r\n-- Don't submit commands\u00a0 in exact command order or parallel processing\r\n-- of indexes\/stats on same object will occur and could block\r\nDECLARE command_cursor CURSOR FOR\r\nSELECT ID\r\nFROM (\r\nSELECT ROW_NUMBER() OVER\r\n(PARTITION BY ObjectName\r\nORDER BY COALESCE(IndexName, StatisticsName), CommandType) AS Ranking,\r\nID\r\nFROM master.dbo.CommandLog\r\nWHERE ID > @MaxID AND ID <= @NewMaxID)\u00a0 AS t\r\nORDER BY Ranking\r\n\r\nOPEN command_cursor\r\n\r\nFETCH NEXT FROM command_cursor\r\nINTO @id\r\n\r\nWHILE @@FETCH_STATUS = 0\r\nBEGIN\r\n\r\n-- Begin a conversation and send a request message\r\nDECLARE @conversation_handle UNIQUEIDENTIFIER;\r\nDECLARE @message_body XML;\r\n\r\nBEGIN TRANSACTION;\r\n\r\nBEGIN DIALOG @conversation_handle\r\nFROM SERVICE [OlaHallengrenMaintenanceTaskService]\r\nTO SERVICE N'OlaHallengrenMaintenanceTaskService'\r\nON CONTRACT [OlaHallengrenMaintenanceTaskContract]\r\nWITH ENCRYPTION = OFF;\r\n\r\nSELECT @message_body = N'<CommandLogID>'+CAST(@id AS NVARCHAR)+N'<\/CommandLogID>';\r\n\r\nSEND ON CONVERSATION @conversation_handle\r\nMESSAGE TYPE [OlaHallengrenMaintenanceTaskMessage]\r\n(@message_body);\r\n\r\nCOMMIT TRANSACTION;\r\n\r\n-- Get the next command to run\r\nFETCH NEXT FROM command_cursor\r\nINTO @id\r\nEND\r\nCLOSE command_cursor;\r\nDEALLOCATE command_cursor;\r\nGO\r\n<\/pre>\n