Skip to content

Commit 032c5e3

Browse files
george.crockerthecrockster
george.crocker
authored and
thecrockster
committed
Only take app lock out at beginning of transaction (#983)
Move sp_getapplock from being prepended to every SQL query to being called once at the beginning of each transaction via a new AcquireAppLockAsync helper method in SqlTriggerUtils. Since the app lock is transaction-scoped, subsequent acquisitions within the same transaction were no-ops that only added noise to query logs. This reduces redundant sp_getapplock calls from ~12 per cycle to 5 (one per transaction). Changes: - Add AcquireAppLockAsync static method to SqlTriggerUtils - Remove {AppLockStatements} from all 13 individual query strings - Call AcquireAppLockAsync once after BeginTransaction in all 5 transaction sites across SqlTableChangeMonitor, SqlTriggerListener, and SqlTriggerMetricsProvider Closes #983
1 parent 750de68 commit 032c5e3

4 files changed

Lines changed: 30 additions & 26 deletions

File tree

‎src/TriggerBinding/SqlTableChangeMonitor.cs‎

Lines changed: 8 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@
1212
using Microsoft.Azure.WebJobs.Extensions.Sql.Telemetry;
1313
using static Microsoft.Azure.WebJobs.Extensions.Sql.Telemetry.Telemetry;
1414
using static Microsoft.Azure.WebJobs.Extensions.Sql.SqlTriggerConstants;
15+
using static Microsoft.Azure.WebJobs.Extensions.Sql.SqlTriggerUtils;
1516
using Microsoft.Azure.WebJobs.Host.Executors;
1617
using Microsoft.Data.SqlClient;
1718
using Microsoft.Extensions.Logging;
@@ -285,6 +286,8 @@ private async Task GetTableChangesAsync(SqlConnection connection, CancellationTo
285286
{
286287
try
287288
{
289+
await AcquireAppLockAsync(connection, transaction, this._logger, token);
290+
288291
// Update the version number stored in the global state table if necessary before using it.
289292
using (SqlCommand updateTablesPreInvocationCommand = this.BuildUpdateTablesPreInvocation(connection, transaction))
290293
{
@@ -525,6 +528,8 @@ private async Task RenewLeasesAsync(SqlConnection connection, CancellationToken
525528
{
526529
try
527530
{
531+
await AcquireAppLockAsync(connection, transaction, this._logger, token);
532+
528533
SqlCommand renewLeasesCommand = this.BuildRenewLeasesCommand(connection, transaction);
529534
if (renewLeasesCommand != null)
530535
{
@@ -634,6 +639,8 @@ private async Task ReleaseLeasesAsync(SqlConnection connection, CancellationToke
634639
{
635640
try
636641
{
642+
await AcquireAppLockAsync(connection, transaction, this._logger, token);
643+
637644
// Release the leases held on "_rowsToRelease".
638645
using (SqlCommand releaseLeasesCommand = this.BuildReleaseLeasesCommand(connection, transaction))
639646
{
@@ -789,8 +796,6 @@ private static SqlChangeOperation GetChangeOperation(IReadOnlyDictionary<string,
789796
private SqlCommand BuildUpdateTablesPreInvocation(SqlConnection connection, SqlTransaction transaction)
790797
{
791798
string updateTablesPreInvocationQuery = $@"
792-
{AppLockStatements}
793-
794799
DECLARE @min_valid_version bigint;
795800
SET @min_valid_version = CHANGE_TRACKING_MIN_VALID_VERSION({this._userTableId});
796801
@@ -834,8 +839,6 @@ private SqlCommand BuildGetChangesCommand(SqlConnection connection, SqlTransacti
834839
// up regardless since we know it should be processed - no need to check the change version.
835840
// Once a row is successfully processed the LeaseExpirationTime column is set to NULL.
836841
string getChangesQuery = $@"
837-
{AppLockStatements}
838-
839842
DECLARE @last_sync_version bigint;
840843
SELECT @last_sync_version = LastSyncVersion
841844
FROM {GlobalStateTableName}
@@ -882,8 +885,6 @@ private async Task<string> GetLeaseLockedOrMaxAttemptRowCountMessage(SqlConnecti
882885
// * NULL LeaseExpirationTime OR LeaseExpirationTime <= Current Time
883886
// * No attempts remaining (Attempt count = Max attempts)
884887
string getLeaseLockedOrMaxAttemptRowCountQuery = $@"
885-
{AppLockStatements}
886-
887888
DECLARE @last_sync_version bigint;
888889
SELECT @last_sync_version = LastSyncVersion
889890
FROM {GlobalStateTableName}
@@ -948,8 +949,6 @@ private SqlCommand BuildAcquireLeasesCommand(SqlConnection connection, SqlTransa
948949
const string rowDataParameter = "@rowData";
949950
// Create the merge query that will either update the rows that already exist or insert a new one if it doesn't exist
950951
string query = $@"
951-
{AppLockStatements}
952-
953952
WITH {acquireLeasesCte} AS ( SELECT * FROM OPENJSON(@rowData) WITH ({string.Join(",", cteColumnDefinitions)}) )
954953
MERGE INTO {this._bracketedLeasesTableName}
955954
AS ExistingData
@@ -989,8 +988,6 @@ private SqlCommand BuildRenewLeasesCommand(SqlConnection connection, SqlTransact
989988
return null;
990989
}
991990
string renewLeasesQuery = $@"
992-
{AppLockStatements}
993-
994991
UPDATE {this._bracketedLeasesTableName}
995992
SET {LeasesTableLeaseExpirationTimeColumnName} = DATEADD(second, {LeaseIntervalInSeconds}, SYSDATETIME())
996993
WHERE {matchCondition};
@@ -1021,9 +1018,7 @@ private SqlCommand BuildReleaseLeasesCommand(SqlConnection connection, SqlTransa
10211018
const string rowDataParameter = "@rowData";
10221019

10231020
string releaseLeasesQuery =
1024-
$@"{AppLockStatements}
1025-
1026-
WITH {releaseLeasesCte} AS ( SELECT * FROM OPENJSON(@rowData) WITH ({string.Join(",", cteColumnDefinitions)}) )
1021+
$@"WITH {releaseLeasesCte} AS ( SELECT * FROM OPENJSON(@rowData) WITH ({string.Join(",", cteColumnDefinitions)}) )
10271022
UPDATE {this._bracketedLeasesTableName}
10281023
SET
10291024
{LeasesTableChangeVersionColumnName} = cte.{SysChangeVersionColumnName},
@@ -1053,8 +1048,6 @@ private SqlCommand BuildUpdateTablesPostInvocation(SqlConnection connection, Sql
10531048
string leasesTableJoinCondition = string.Join(" AND ", this._primaryKeyColumns.Select(col => $"c.{col.name.AsBracketQuotedString()} = l.{col.name.AsBracketQuotedString()}"));
10541049

10551050
string updateTablesPostInvocationQuery = $@"
1056-
{AppLockStatements}
1057-
10581051
DECLARE @current_last_sync_version bigint;
10591052
SELECT @current_last_sync_version = LastSyncVersion
10601053
FROM {GlobalStateTableName}

‎src/TriggerBinding/SqlTriggerListener.cs‎

Lines changed: 2 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -143,6 +143,8 @@ public async Task StartAsync(CancellationToken cancellationToken)
143143
long createdSchemaDurationMs = 0L, createGlobalStateTableDurationMs = 0L, insertGlobalStateTableRowDurationMs = 0L, createLeasesTableDurationMs = 0L;
144144
using (SqlTransaction transaction = connection.BeginTransaction(System.Data.IsolationLevel.RepeatableRead))
145145
{
146+
await AcquireAppLockAsync(connection, transaction, this._logger, cancellationToken);
147+
146148
createdSchemaDurationMs = await this.CreateSchemaAsync(connection, transaction, cancellationToken);
147149
createGlobalStateTableDurationMs = await this.CreateGlobalStateTableAsync(connection, transaction, cancellationToken);
148150
insertGlobalStateTableRowDurationMs = await this.InsertGlobalStateTableRowAsync(connection, transaction, userTableId, cancellationToken);
@@ -283,8 +285,6 @@ FROM sys.columns AS c
283285
private async Task<long> CreateSchemaAsync(SqlConnection connection, SqlTransaction transaction, CancellationToken cancellationToken)
284286
{
285287
string createSchemaQuery = $@"
286-
{AppLockStatements}
287-
288288
IF SCHEMA_ID(N'{SchemaName}') IS NULL
289289
EXEC ('CREATE SCHEMA {SchemaName}');
290290
";
@@ -328,8 +328,6 @@ IF SCHEMA_ID(N'{SchemaName}') IS NULL
328328
private async Task<long> CreateGlobalStateTableAsync(SqlConnection connection, SqlTransaction transaction, CancellationToken cancellationToken)
329329
{
330330
string createGlobalStateTableQuery = $@"
331-
{AppLockStatements}
332-
333331
IF OBJECT_ID(N'{GlobalStateTableName}', 'U') IS NULL
334332
CREATE TABLE {GlobalStateTableName} (
335333
UserFunctionID char(16) NOT NULL,
@@ -401,7 +399,6 @@ private async Task<long> InsertGlobalStateTableRowAsync(SqlConnection connection
401399
}
402400

403401
string insertRowGlobalStateTableQuery = $@"
404-
{AppLockStatements}
405402
-- For back compatibility copy the lastSyncVersion from _hostIdFunctionId if it exists.
406403
IF NOT EXISTS (
407404
SELECT * FROM {GlobalStateTableName}
@@ -456,8 +453,6 @@ private async Task<long> CreateLeasesTableAsync(
456453
// we're actually using the WEBSITE_SITE_NAME one (e.g. leasesTableName is different)
457454
bool shouldMigrateOldLeasesTable = !string.IsNullOrEmpty(oldLeasesTableName) && oldLeasesTableName != leasesTableName;
458455
string createLeasesTableQuery = shouldMigrateOldLeasesTable ? $@"
459-
{AppLockStatements}
460-
461456
IF OBJECT_ID(N'{leasesTableName}', 'U') IS NULL
462457
BEGIN
463458
CREATE TABLE {leasesTableName} (
@@ -479,8 +474,6 @@ INSERT INTO {leasesTableName}
479474
End
480475
" :
481476
$@"
482-
{AppLockStatements}
483-
484477
IF OBJECT_ID(N'{leasesTableName}', 'U') IS NULL
485478
CREATE TABLE {leasesTableName} (
486479
{primaryKeysWithTypes},

‎src/TriggerBinding/SqlTriggerMetricsProvider.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,6 +63,8 @@ private async Task<long> GetUnprocessedChangeCountAsync()
6363
{
6464
try
6565
{
66+
await AcquireAppLockAsync(connection, transaction, this._logger, CancellationToken.None);
67+
6668
using (SqlCommand getUnprocessedChangesCommand = this.BuildGetUnprocessedChangesCommand(connection, transaction, primaryKeyColumns, userTableId))
6769
{
6870
var commandSw = Stopwatch.StartNew();
@@ -102,8 +104,6 @@ private SqlCommand BuildGetUnprocessedChangesCommand(SqlConnection connection, S
102104
string leasesTableJoinCondition = string.Join(" AND ", primaryKeyColumns.Select(col => $"c.{col.name.AsBracketQuotedString()} = l.{col.name.AsBracketQuotedString()}"));
103105
string bracketedLeasesTableName = GetBracketedLeasesTableName(this._userDefinedLeasesTableName, this._userFunctionId, userTableId);
104106
string getUnprocessedChangesQuery = $@"
105-
{AppLockStatements}
106-
107107
DECLARE @last_sync_version bigint;
108108
SELECT @last_sync_version = LastSyncVersion
109109
FROM {GlobalStateTableName}

‎src/TriggerBinding/SqlTriggerUtils.cs‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,24 @@ namespace Microsoft.Azure.WebJobs.Extensions.Sql
1616
public static class SqlTriggerUtils
1717
{
1818

19+
/// <summary>
20+
/// Acquires an exclusive application lock on the transaction to prevent deadlocks.
21+
/// This should be called once at the beginning of each transaction rather than
22+
/// being included in every individual query, since the lock is transaction-scoped
23+
/// and subsequent acquisitions within the same transaction are no-ops.
24+
/// </summary>
25+
/// <param name="connection">The SQL connection</param>
26+
/// <param name="transaction">The transaction to acquire the lock on</param>
27+
/// <param name="logger">Logger for logging the command</param>
28+
/// <param name="cancellationToken">Cancellation token</param>
29+
public static async Task AcquireAppLockAsync(SqlConnection connection, SqlTransaction transaction, ILogger logger, CancellationToken cancellationToken)
30+
{
31+
using (var command = new SqlCommand(AppLockStatements, connection, transaction))
32+
{
33+
await command.ExecuteNonQueryAsyncWithLogging(logger, cancellationToken, true);
34+
}
35+
}
36+
1937
/// <summary>
2038
/// Gets the names and types of primary key columns of the user table.
2139
/// </summary>

0 commit comments

Comments
 (0)