Fixed bugs, refactored and polished distributed lock service.

- Added machine and thread scoped locking.
- Added support for blocking acquisitions.
- Added thread synchronization when creating and committing nested transactions for lock record creation and updates.
This commit is contained in:
Sipke Schoorstra
2015-08-21 15:22:32 +01:00
parent 61f1c3ae00
commit 6c712a9840
18 changed files with 507 additions and 256 deletions

View File

@@ -269,7 +269,7 @@
<Compile Include="Localization\DateTimePartsTests.cs" /> <Compile Include="Localization\DateTimePartsTests.cs" />
<Compile Include="Localization\DefaultDateLocalizationServicesTests.cs" /> <Compile Include="Localization\DefaultDateLocalizationServicesTests.cs" />
<Compile Include="Localization\DefaultDateFormatterTests.cs" /> <Compile Include="Localization\DefaultDateFormatterTests.cs" />
<Compile Include="Stubs\StubDistributedLock.cs" /> <Compile Include="Stubs\StubThreadProvider.cs" />
<Compile Include="Stubs\StubMachineNameProvider.cs" /> <Compile Include="Stubs\StubMachineNameProvider.cs" />
<Compile Include="Stubs\StubCultureSelector.cs" /> <Compile Include="Stubs\StubCultureSelector.cs" />
<Compile Include="Localization\TestHelpers.cs" /> <Compile Include="Localization\TestHelpers.cs" />

View File

@@ -1,15 +0,0 @@
using System;
using Orchard.Tasks.Locking.Services;
namespace Orchard.Tests.Stubs {
public class StubDistributedLock : IDistributedLock {
public bool IsDisposed { get; private set; }
public void Dispose() {
IsDisposed = true;
}
public int Id { get; set; }
public string Name { get; set; }
}
}

View File

@@ -0,0 +1,15 @@
using Orchard.Environment;
namespace Orchard.Tests.Stubs {
public class StubThreadProvider : IThreadProvider {
public StubThreadProvider() {
ManagedThreadId = 1;
}
public int ManagedThreadId { get; set; }
public int GetCurrentThreadId() {
return ManagedThreadId;
}
}
}

View File

@@ -1,6 +1,5 @@
using System; using System;
using System.Collections.Generic; using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks; using System.Threading.Tasks;
using Autofac; using Autofac;
using NUnit.Framework; using NUnit.Framework;
@@ -16,103 +15,208 @@ namespace Orchard.Tests.Tasks {
public class DistributedLockServiceTests : DatabaseEnabledTestsBase { public class DistributedLockServiceTests : DatabaseEnabledTestsBase {
private const string LockName = "Orchard Test Lock"; private const string LockName = "Orchard Test Lock";
private DistributedLockService _distributedLockService; private DistributedLockService _distributedLockService;
private StubMachineNameProvider _stubMachineNameProvider; private StubMachineNameProvider _machineNameProvider;
private IRepository<LockRecord> _lockRepository; private StubThreadProvider _threadProvider;
private IRepository<DistributedLockRecord> _distributedLockRepository;
protected override IEnumerable<Type> DatabaseTypes protected override IEnumerable<Type> DatabaseTypes {
{ get { yield return typeof(DistributedLockRecord); }
get { yield return typeof (LockRecord); }
} }
public override void Register(ContainerBuilder builder) { public override void Register(ContainerBuilder builder) {
builder.RegisterType<StubClock>().As<IClock>(); builder.RegisterType<StubClock>().As<IClock>();
//builder.RegisterType<StubWorkContextAccessor>().As<IWorkContextAccessor>();
builder.RegisterType<StubMachineNameProvider>().As<IMachineNameProvider>().SingleInstance(); builder.RegisterType<StubMachineNameProvider>().As<IMachineNameProvider>().SingleInstance();
builder.RegisterType<StubDistributedLock>().As<IDistributedLock>(); builder.RegisterType<StubThreadProvider>().As<IThreadProvider>().SingleInstance();
builder.RegisterType<DistributedLockService>().AsSelf(); builder.RegisterType<DistributedLockService>().AsSelf();
builder.RegisterInstance(new Work<IDistributedLock>(resolve => _container.Resolve<IDistributedLock>())).AsSelf();
} }
public override void Init() { public override void Init() {
base.Init(); base.Init();
_distributedLockService = _container.Resolve<DistributedLockService>(); _distributedLockService = _container.Resolve<DistributedLockService>();
_stubMachineNameProvider = (StubMachineNameProvider)_container.Resolve<IMachineNameProvider>(); _machineNameProvider = (StubMachineNameProvider)_container.Resolve<IMachineNameProvider>();
_lockRepository = _container.Resolve<IRepository<LockRecord>>(); _threadProvider = (StubThreadProvider)_container.Resolve<IThreadProvider>();
_distributedLockRepository = _container.Resolve<IRepository<DistributedLockRecord>>();
} }
[Test] [Test]
public void AcquiringLockSucceeds() { public void TryAcquiringLockSucceeds() {
IDistributedLock @lock; DistributedLock @lock;
var lockAcquired = _distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); var lockAcquired = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
Assert.That(lockAcquired, Is.True); Assert.That(lockAcquired, Is.True);
} }
[Test] [Test]
public void AcquiringLockTwiceOnSameMachineSucceeds() { public void TryAcquiringLockTwiceOnSameMachineSucceeds() {
IDistributedLock @lock; DistributedLock @lock;
var attempt1 = _distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); var attempt1 = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
var attempt2 = _distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); var attempt2 = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
Assert.That(attempt1, Is.True); Assert.That(attempt1, Is.True);
Assert.That(attempt2, Is.True); Assert.That(attempt2, Is.True);
} }
[Test] [Test]
public void AcquiringLockTwiceOnSameMachineIncreasesRefCountTwice() { public void TryAcquiringLockTwiceOnSameMachineIncreasesLockCountTwice() {
IDistributedLock @lock; DistributedLock @lock;
_distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
_distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
var lockRecord = _lockRepository.Get(@lock.Id); var lockId = Int32.Parse(@lock.Id);
Assert.That(lockRecord.ReferenceCount, Is.EqualTo(2)); var lockRecord = _distributedLockRepository.Get(lockId);
Assert.That(lockRecord.Count, Is.EqualTo(2));
} }
[Test] [Test]
public void DisposingLockWillDecreaseRefCount() { public void ReleasingLockDecreasesLockCount() {
IDistributedLock @lock; DistributedLock @lock;
_distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
_distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
var lockRecord = _lockRepository.Get(@lock.Id);
_distributedLockService.DisposeLock(@lock); var lockId = Int32.Parse(@lock.Id);
Assert.That(lockRecord.ReferenceCount, Is.EqualTo(1)); var lockRecord = _distributedLockRepository.Get(lockId);
_distributedLockService.DisposeLock(@lock); _distributedLockService.ReleaseLock(@lock);
Assert.That(lockRecord.ReferenceCount, Is.EqualTo(0)); Assert.That(lockRecord.Count, Is.EqualTo(1));
_distributedLockService.ReleaseLock(@lock);
Assert.That(lockRecord.Count, Is.EqualTo(0));
} }
[Test] [Test]
public void AcquiringLockTwiceFails() { public void TryAcquiringLockTwiceFails() {
IDistributedLock @lock; DistributedLock @lock;
_stubMachineNameProvider.MachineName = "Orchard Test Machine 1"; _machineNameProvider.MachineName = "Orchard Test Machine 1";
var attempt1 = _distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); var attempt1 = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
_stubMachineNameProvider.MachineName = "Orchard Test Machine 2"; _machineNameProvider.MachineName = "Orchard Test Machine 2";
var attempt2 = _distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock); var attempt2 = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
Assert.That(attempt1, Is.True); Assert.That(attempt1, Is.True);
Assert.That(attempt2, Is.False); Assert.That(attempt2, Is.False);
} }
[Test] [Test]
public void MultipleSimultaneousAcquisitionsShouldAllowOneLock() { public void TryAcquiringNonExpiredActiveLockFails() {
var attempts = new List<bool>(); DistributedLock @lock;
CreateNonExpiredActiveLock("Other Machine", threadId: null);
var success = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromHours(1), null, out @lock);
Assert.That(success, Is.False);
}
[Test]
public void TryAcquiringNonExpiredButInactiveLockSucceeds() {
DistributedLock @lock;
CreateNonExpiredButInactiveLock("Other Machine", threadId: null);
var success = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromHours(1), null, out @lock);
Assert.That(success, Is.True);
}
[Test]
public void TryAcquiringExpiredButActiveLockSucceeds() {
DistributedLock @lock;
CreateExpiredButActiveLock("Other Machine", threadId: null);
var success = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromHours(1), null, out @lock);
Assert.That(success, Is.True);
}
[Test]
public void TryAcquiringNonExpiredAndActiveLockFromCurrentOwnerSucceeds() {
DistributedLock @lock;
CreateNonExpiredActiveLock(GetMachineName(), threadId: null);
var success = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromHours(1), null, out @lock);
Assert.That(success, Is.True);
}
[Test]
public void AcquiringNonExpiredAndActiveLockFromDifferentOwnerThrowsTimeoutException() {
CreateNonExpiredActiveLock("Other Machine", threadId: null);
Assert.Throws<TimeoutException>(() => _distributedLockService.AcquireLockForMachine(LockName, TimeSpan.FromHours(1), TimeSpan.Zero));
}
[Test]
public void MultipleAcquisitionsFromDifferentMachinesShouldFail() {
DistributedLock @lock;
_machineNameProvider.MachineName = "Orchard Test Machine 1";
var attempt1 = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
_machineNameProvider.MachineName = "Orchard Test Machine 2";
var attempt2 = _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock);
Assert.That(attempt1, Is.True);
Assert.That(attempt2, Is.False);
}
[Test]
public void MultithreadedAcquisitionsShouldNotCauseTransactionErrors() {
var tasks = new List<Task>(); var tasks = new List<Task>();
foreach (var index in Enumerable.Range(0, 20)) { for (var i = 0; i < 10; i++) {
var task = Task.Factory.StartNew(() => { var task = Task.Factory.StartNew(() => {
IDistributedLock @lock; DistributedLock @lock;
_stubMachineNameProvider.MachineName = "Orchard Test Machine " + (index + 1); Assert.DoesNotThrow(() => _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromHours(1), null, out @lock));
var attempt = _distributedLockService.TryAcquireLock(LockName, TimeSpan.FromSeconds(60), TimeSpan.Zero, out @lock);
attempts.Add(attempt);
}); });
tasks.Add(task); tasks.Add(task);
} }
Task.WaitAll(tasks.ToArray()); Task.WaitAll(tasks.ToArray());
Assert.That(attempts.Count(x => x == true), Is.EqualTo(1)); }
[Test]
public void MixedScopeAcquisitionsShouldThrow() {
DistributedLock @lock;
Assert.DoesNotThrow(() => _distributedLockService.TryAcquireLockForMachine(LockName, TimeSpan.FromSeconds(60), null, out @lock));
Assert.Throws<InvalidOperationException>(() => _distributedLockService.TryAcquireLockForThread(LockName, TimeSpan.FromSeconds(60), null, out @lock));
}
[Test]
public void TryAcquireActiveLockWithNullTimeoutReturnsFalseImmediately() {
CreateNonExpiredActiveLock("Other Machine", null);
DistributedLock @lock;
var acquired = _distributedLockService.TryAcquireLockForThread(LockName, TimeSpan.FromMinutes(1), null, out @lock);
Assert.That(acquired, Is.False);
}
private DistributedLockRecord CreateLockRecord(int count, DateTime createdUtc, DateTime validUntilUtc, string machineName, int? threadId) {
var record = new DistributedLockRecord {
Name = LockName,
Count = count,
CreatedUtc = createdUtc,
ValidUntilUtc = validUntilUtc,
MachineName = machineName,
ThreadId = threadId
};
_distributedLockRepository.Create(record);
return record;
}
private DistributedLockRecord CreateNonExpiredActiveLock(string machineName, int? threadId) {
var now = _clock.UtcNow;
return CreateLockRecord(1, now, now + TimeSpan.FromHours(1), machineName, threadId);
}
private DistributedLockRecord CreateNonExpiredButInactiveLock(string machineName, int? threadId) {
var now = _clock.UtcNow;
return CreateLockRecord(0, now, now + TimeSpan.FromHours(1), machineName, threadId);
}
private DistributedLockRecord CreateExpiredButActiveLock(string machineName, int? threadId) {
var now = _clock.UtcNow;
return CreateLockRecord(1, now, now - TimeSpan.FromHours(1), machineName, threadId);
}
private string GetMachineName() {
return _machineNameProvider.GetMachineName();
}
private int GetThreadId() {
return _threadProvider.GetCurrentThreadId();
} }
} }
} }

View File

@@ -1,5 +1,4 @@
using System; using Autofac;
using Autofac;
using Moq; using Moq;
using NUnit.Framework; using NUnit.Framework;
using Orchard.Tasks.Locking.Services; using Orchard.Tasks.Locking.Services;
@@ -8,9 +7,9 @@ namespace Orchard.Tests.Tasks {
[TestFixture] [TestFixture]
public class LockTests : ContainerTestBase { public class LockTests : ContainerTestBase {
private const string LockName = "Orchard Test Lock"; private const string LockName = "Orchard Test Lock";
private const int LockId = 1; private const string LockId = "1";
private Mock<IDistributedLockService> _distributedLockServiceMock; private Mock<IDistributedLockService> _distributedLockServiceMock;
private Lock _lock; private DistributedLock _lock;
protected override void Register(ContainerBuilder builder) { protected override void Register(ContainerBuilder builder) {
_distributedLockServiceMock = new Mock<IDistributedLockService>(); _distributedLockServiceMock = new Mock<IDistributedLockService>();
@@ -18,14 +17,14 @@ namespace Orchard.Tests.Tasks {
} }
protected override void Resolve(ILifetimeScope container) { protected override void Resolve(ILifetimeScope container) {
_lock = new Lock(_distributedLockServiceMock.Object, LockName, LockId); _lock = DistributedLock.ForMachine(_distributedLockServiceMock.Object, LockName, "Orchard Test Machine", LockId);
} }
[Test] [Test]
public void DisposeInvokesDistributedLockServiceDisposeLock() { public void DisposeInvokesDistributedLockServiceDisposeLock() {
_lock.Dispose(); _lock.Dispose();
_distributedLockServiceMock.Verify(service => service.DisposeLock(_lock), Times.Exactly(1)); _distributedLockServiceMock.Verify(service => service.ReleaseLock(_lock), Times.Exactly(1));
} }
[Test] [Test]
@@ -34,7 +33,7 @@ namespace Orchard.Tests.Tasks {
_lock.Dispose(); _lock.Dispose();
_lock.Dispose(); _lock.Dispose();
_distributedLockServiceMock.Verify(service => service.DisposeLock(_lock), Times.Exactly(1)); _distributedLockServiceMock.Verify(service => service.ReleaseLock(_lock), Times.Exactly(1));
} }
} }
} }

View File

@@ -29,27 +29,27 @@ namespace Orchard.Data.Migration {
public ILogger Logger { get; set; } public ILogger Logger { get; set; }
public void Activated() { public void Activated() {
using(var @lock = _distributedLockService.AcquireLock(GetType().FullName, TimeSpan.FromMinutes(30), TimeSpan.FromMilliseconds(250))) { DistributedLock @lock;
if (@lock == null) if(_distributedLockService.TryAcquireLockForMachine(GetType().FullName, TimeSpan.FromMinutes(30), TimeSpan.FromMilliseconds(250), out @lock)) {
return; using (@lock) {
// Let's make sure that the basic set of features is enabled. If there are any that are not enabled, then let's enable them first.
var theseFeaturesShouldAlwaysBeActive = new[] {
"Common", "Containers", "Contents", "Dashboard", "Feeds", "Navigation", "Scheduling", "Settings", "Shapes", "Title"
};
// Let's make sure that the basic set of features is enabled. If there are any that are not enabled, then let's enable them first. var enabledFeatures = _featureManager.GetEnabledFeatures().Select(f => f.Id).ToList();
var theseFeaturesShouldAlwaysBeActive = new[] { var featuresToEnable = theseFeaturesShouldAlwaysBeActive.Where(shouldBeActive => !enabledFeatures.Contains(shouldBeActive)).ToList();
"Common", "Containers", "Contents", "Dashboard", "Feeds", "Navigation", "Scheduling", "Settings", "Shapes", "Title" if (featuresToEnable.Any()) {
}; _featureManager.EnableFeatures(featuresToEnable, true);
var enabledFeatures = _featureManager.GetEnabledFeatures().Select(f => f.Id).ToList();
var featuresToEnable = theseFeaturesShouldAlwaysBeActive.Where(shouldBeActive => !enabledFeatures.Contains(shouldBeActive)).ToList();
if (featuresToEnable.Any()) {
_featureManager.EnableFeatures(featuresToEnable, true);
}
foreach (var feature in _dataMigrationManager.GetFeaturesThatNeedUpdate()) {
try {
_dataMigrationManager.Update(feature);
} }
catch (Exception e) {
Logger.Error("Could not run migrations automatically on " + feature, e); foreach (var feature in _dataMigrationManager.GetFeaturesThatNeedUpdate()) {
try {
_dataMigrationManager.Update(feature);
}
catch (Exception ex) {
Logger.Error(ex, "Could not run migrations automatically on {0}.", feature);
}
} }
} }
} }

View File

@@ -1,7 +1,7 @@
namespace Orchard.Environment { namespace Orchard.Environment {
/// <summary> /// <summary>
/// Describes a service which the name of the machine running the application. /// Describes a service which returns the name of the machine running the application.
/// </summary> /// </summary>
public interface IMachineNameProvider { public interface IMachineNameProvider {

View File

@@ -0,0 +1,13 @@
namespace Orchard.Environment {
/// <summary>
/// Describes a service which returns the managed thread ID of the current thread.
/// </summary>
public interface IThreadProvider {
/// <summary>
/// Returns the managed thread ID of the current thread.
/// </summary>
int GetCurrentThreadId();
}
}

View File

@@ -67,6 +67,7 @@ namespace Orchard.Environment {
builder.RegisterType<ViewsBackgroundCompilation>().As<IViewsBackgroundCompilation>().SingleInstance(); builder.RegisterType<ViewsBackgroundCompilation>().As<IViewsBackgroundCompilation>().SingleInstance();
builder.RegisterType<DefaultExceptionPolicy>().As<IExceptionPolicy>().SingleInstance(); builder.RegisterType<DefaultExceptionPolicy>().As<IExceptionPolicy>().SingleInstance();
builder.RegisterType<DefaultCriticalErrorProvider>().As<ICriticalErrorProvider>().SingleInstance(); builder.RegisterType<DefaultCriticalErrorProvider>().As<ICriticalErrorProvider>().SingleInstance();
builder.RegisterType<DefaultCriticalErrorProvider>().As<IThreadProvider>().SingleInstance();
//builder.RegisterType<RazorTemplateCache>().As<IRazorTemplateProvider>().SingleInstance(); //builder.RegisterType<RazorTemplateCache>().As<IRazorTemplateProvider>().SingleInstance();
RegisterVolatileProvider<WebSiteFolder, IWebSiteFolder>(builder); RegisterVolatileProvider<WebSiteFolder, IWebSiteFolder>(builder);

View File

@@ -0,0 +1,9 @@
using System.Threading;
namespace Orchard.Environment {
public class ThreadProvider : IThreadProvider {
public int GetCurrentThreadId() {
return Thread.CurrentThread.ManagedThreadId;
}
}
}

View File

@@ -151,6 +151,8 @@
<ItemGroup> <ItemGroup>
<Compile Include="Environment\Extensions\Helpers\ExtensionManagerExtensions.cs" /> <Compile Include="Environment\Extensions\Helpers\ExtensionManagerExtensions.cs" />
<Compile Include="Environment\Extensions\Models\LifecycleStatus.cs" /> <Compile Include="Environment\Extensions\Models\LifecycleStatus.cs" />
<Compile Include="Environment\IThreadProvider.cs" />
<Compile Include="Environment\ThreadProvider.cs" />
<Compile Include="Mvc\Updater.cs" /> <Compile Include="Mvc\Updater.cs" />
<Compile Include="Recipes\Models\ConfigurationContext.cs" /> <Compile Include="Recipes\Models\ConfigurationContext.cs" />
<Compile Include="Recipes\Models\RecipeBuilderStepConfigurationContext.cs" /> <Compile Include="Recipes\Models\RecipeBuilderStepConfigurationContext.cs" />
@@ -398,11 +400,10 @@
<Compile Include="Settings\CurrentSiteWorkContext.cs" /> <Compile Include="Settings\CurrentSiteWorkContext.cs" />
<Compile Include="Settings\ResourceDebugMode.cs" /> <Compile Include="Settings\ResourceDebugMode.cs" />
<Compile Include="Tasks\Locking\Migrations\FrameworkMigrations.cs" /> <Compile Include="Tasks\Locking\Migrations\FrameworkMigrations.cs" />
<Compile Include="Tasks\Locking\Services\Lock.cs" /> <Compile Include="Tasks\Locking\Services\DistributedLock.cs" />
<Compile Include="Tasks\Locking\Services\IDistributedLockService.cs" /> <Compile Include="Tasks\Locking\Services\IDistributedLockService.cs" />
<Compile Include="Tasks\Locking\Services\IDistributedLock.cs" />
<Compile Include="Tasks\Locking\Services\DistributedLockService.cs" /> <Compile Include="Tasks\Locking\Services\DistributedLockService.cs" />
<Compile Include="Tasks\Locking\Records\LockRecord.cs" /> <Compile Include="Tasks\Locking\Records\DistributedLockRecord.cs" />
<Compile Include="Themes\CurrentThemeWorkContext.cs" /> <Compile Include="Themes\CurrentThemeWorkContext.cs" />
<Compile Include="Themes\ThemeManager.cs" /> <Compile Include="Themes\ThemeManager.cs" />
<Compile Include="Time\CurrentTimeZoneWorkContext.cs" /> <Compile Include="Time\CurrentTimeZoneWorkContext.cs" />

View File

@@ -5,11 +5,12 @@ namespace Orchard.Tasks.Locking.Migrations {
public class FrameworkMigrations : DataMigrationImpl { public class FrameworkMigrations : DataMigrationImpl {
public int Create() { public int Create() {
SchemaBuilder.CreateTable("LockRecord", table => table SchemaBuilder.CreateTable("DistributedLockRecord", table => table
.Column<int>("Id", column => column.PrimaryKey().Identity()) .Column<int>("Id", column => column.PrimaryKey().Identity())
.Column<string>("Name", column => column.NotNull().WithLength(256)) .Column<string>("Name", column => column.NotNull().WithLength(256))
.Column<string>("Owner", column => column.WithLength(256)) .Column<string>("MachineName", column => column.WithLength(256))
.Column<int>("ReferenceCount") .Column<int>("ThreadId", column => column.Nullable())
.Column<int>("Count")
.Column<DateTime>("CreatedUtc") .Column<DateTime>("CreatedUtc")
.Column<DateTime>("ValidUntilUtc")); .Column<DateTime>("ValidUntilUtc"));

View File

@@ -1,11 +1,12 @@
using System; using System;
namespace Orchard.Tasks.Locking.Records { namespace Orchard.Tasks.Locking.Records {
public class LockRecord { public class DistributedLockRecord {
public virtual int Id { get; set; } public virtual int Id { get; set; }
public virtual string Name { get; set; } public virtual string Name { get; set; }
public virtual string Owner { get; set; } public virtual string MachineName { get; set; }
public virtual int ReferenceCount { get; set; } public virtual int? ThreadId { get; set; }
public virtual int Count { get; set; }
public virtual DateTime CreatedUtc { get; set; } public virtual DateTime CreatedUtc { get; set; }
public virtual DateTime ValidUntilUtc { get; set; } public virtual DateTime ValidUntilUtc { get; set; }
} }

View File

@@ -0,0 +1,46 @@
using System;
using System.Threading;
namespace Orchard.Tasks.Locking.Services {
/// <summary>
/// Represents a distributed lock. />
/// </summary>
public class DistributedLock : IDisposable {
private IDistributedLockService _service;
private int _isDisposed;
private DistributedLock() {
}
public string Id { get; private set; }
public string Name { get; private set; }
public string MachineName { get; private set; }
public int? ThreadId { get; private set; }
// This will be called at least and at the latest by the IoC container when the request ends.
public void Dispose() {
if(Interlocked.CompareExchange(ref _isDisposed, 1, 0) == 0)
_service.ReleaseLock(this);
}
public static DistributedLock ForMachine(IDistributedLockService service, string name, string machineName, string lockId) {
return new DistributedLock {
_service = service,
Name = name,
MachineName = machineName,
Id = lockId
};
}
public static DistributedLock ForThread(IDistributedLockService service, string name, string machineName, int threadId, string lockId) {
return new DistributedLock {
_service = service,
Name = name,
MachineName = machineName,
ThreadId = threadId,
Id = lockId
};
}
}
}

View File

@@ -1,7 +1,6 @@
using System; using System;
using System.Data; using System.Data;
using System.Linq; using System.Linq;
using System.Threading;
using System.Threading.Tasks; using System.Threading.Tasks;
using Autofac; using Autofac;
using Orchard.Data; using Orchard.Data;
@@ -13,137 +12,189 @@ using Orchard.Tasks.Locking.Records;
namespace Orchard.Tasks.Locking.Services { namespace Orchard.Tasks.Locking.Services {
public class DistributedLockService : IDistributedLockService { public class DistributedLockService : Component, IDistributedLockService {
private readonly IMachineNameProvider _machineNameProvider; private readonly IMachineNameProvider _machineNameProvider;
private readonly ILifetimeScope _lifetimeScope; private readonly ILifetimeScope _lifetimeScope;
private readonly IClock _clock; private readonly IClock _clock;
private readonly object _semaphore = new object(); private readonly object _transactionManagerLock = new object();
private readonly IThreadProvider _threadProvider;
public DistributedLockService(IMachineNameProvider machineNameProvider, ILifetimeScope lifetimeScope, IClock clock) { public DistributedLockService(IMachineNameProvider machineNameProvider, IThreadProvider threadProvider, ILifetimeScope lifetimeScope, IClock clock) {
_machineNameProvider = machineNameProvider; _machineNameProvider = machineNameProvider;
_lifetimeScope = lifetimeScope; _lifetimeScope = lifetimeScope;
_clock = clock; _clock = clock;
Logger = NullLogger.Instance; _threadProvider = threadProvider;
} }
public ILogger Logger { get; set; } public bool TryAcquireLockForMachine(string name, TimeSpan maxValidFor, TimeSpan? timeout, out DistributedLock @lock) {
return TryAcquireLock(name, maxValidFor, timeout, GetMachineName(), null, out @lock);
}
public bool TryAcquireLock(string name, TimeSpan maxLifetime, TimeSpan timeout, out IDistributedLock @lock) { public DistributedLock AcquireLockForMachine(string name, TimeSpan maxValidFor, TimeSpan? timeout) {
lock (_semaphore) { return AcquireLock(name, maxValidFor, timeout, GetMachineName(), null);
@lock = default(IDistributedLock); }
public bool TryAcquireLockForThread(string name, TimeSpan maxValidFor, TimeSpan? timeout, out DistributedLock @lock) {
return TryAcquireLock(name, maxValidFor, timeout, GetMachineName(), GetThreadId(), out @lock);
}
public DistributedLock AcquireLockForThread(string name, TimeSpan maxValidFor, TimeSpan? timeout) {
return AcquireLock(name, maxValidFor, timeout, GetMachineName(), GetThreadId());
}
public void ReleaseLock(DistributedLock @lock) {
lock (_transactionManagerLock) {
var childLifetimeScope = CreateChildLifetimeScope(@lock.Name);
try { try {
var waitedTime = TimeSpan.Zero; var repository = childLifetimeScope.Resolve<IRepository<DistributedLockRecord>>();
var waitTime = TimeSpan.FromMilliseconds(timeout.TotalMilliseconds / 10); var transactionManager = childLifetimeScope.Resolve<ITransactionManager>();
bool acquired; transactionManager.RequireNew(IsolationLevel.ReadCommitted);
var lockId = Int32.Parse(@lock.Id);
var record = repository.Get(lockId);
while (!(acquired = TryAcquireLockRecord(name, maxLifetime, out @lock)) && waitedTime < timeout) { if (record == null)
Task.Delay(timeout).ContinueWith(t => { throw new ObjectDisposedException("@lock", "No lock record could be found for the specified lock to be released.");
waitedTime += waitTime;
}).Wait();
}
if (acquired) { if (record.Count <= 0)
Logger.Debug("Successfully acquired a lock named {0}.", name); throw new ObjectDisposedException("@lock", "The specified lock has already been released.");
return true;
} record.Count--;
} }
catch (Exception ex) { catch (Exception ex) {
Logger.Error(ex, "Error while trying to acquire a lock named {0}.", name); if (ex.IsFatal()) throw;
throw; Logger.Error(ex, "An non-fatal error occurred while trying to dispose a distributed lock with name '{0}' and ID {1}.", @lock.Name, @lock.Id);
}
finally {
childLifetimeScope.Dispose();
} }
Logger.Debug("Could not acquire a lock named {0}.", name);
return false;
} }
} }
public IDistributedLock AcquireLock(string name, TimeSpan maxLifetime, TimeSpan timeout) { private bool TryAcquireLock(string name, TimeSpan maxValidFor, TimeSpan? timeout, string machineName, int? threadId, out DistributedLock @lock) {
IDistributedLock lockResult; @lock = AcquireLockInternal(name, maxValidFor, machineName, threadId, timeout.GetValueOrDefault());
return TryAcquireLock(name, maxLifetime, timeout, out lockResult) ? lockResult : null; if (@lock != null)
}
public void DisposeLock(IDistributedLock @lock) {
var childLifetimeScope = CreateChildLifetimeScope(@lock.Name);
try {
var repository = childLifetimeScope.Resolve<IRepository<LockRecord>>();
var transactionManager = childLifetimeScope.Resolve<ITransactionManager>();
transactionManager.RequireNew(IsolationLevel.ReadCommitted);
var record = repository.Get(@lock.Id);
if (record != null) {
if (record.ReferenceCount > 0)
record.ReferenceCount--;
}
}
catch (Exception ex) {
if (ex.IsFatal()) throw;
Logger.Error(ex, "An non-fatal error occurred while trying to dispose a distributed lock with name {0} and ID {1}.", @lock.Name, @lock.Id);
}
finally {
childLifetimeScope.Dispose();
}
}
protected virtual bool TryAcquireLockRecord(string name, TimeSpan maxLifetime, out IDistributedLock @lock) {
@lock = null;
var childLifetimeScope = CreateChildLifetimeScope(name);
try {
var transactionManager = childLifetimeScope.Resolve<ITransactionManager>();
transactionManager.RequireNew(IsolationLevel.ReadCommitted);
// This way we can create a nested transaction scope instead of having the unwanted effect
// of manipulating the transaction of the caller.
var repository = childLifetimeScope.Resolve<IRepository<LockRecord>>();
// Find an existing, active lock, if any.
var record = repository.Table.FirstOrDefault(x => x.Name == name && (x.ValidUntilUtc >= _clock.UtcNow || x.ReferenceCount > 0));
// The current owner name (based on machine name and current thread ID).
var currentOwnerName = GetOwnerName();
var canAcquireLock = false;
// Check if there's already an active lock.
if (record != null) {
// Check if the owner of the lock is the one trying to acquire the lock.
if (record.Owner == currentOwnerName) {
record.ReferenceCount++;
canAcquireLock = true;
}
}
else {
// No one has an active lock yet, so good to go.
record = new LockRecord {
Name = name,
Owner = currentOwnerName,
ReferenceCount = 1,
CreatedUtc = _clock.UtcNow,
ValidUntilUtc = _clock.UtcNow + maxLifetime
};
repository.Create(record);
repository.Flush();
canAcquireLock = true;
}
if (!canAcquireLock)
return false;
@lock = new Lock(this, name, record.Id);
return true; return true;
Logger.Debug("Could not acquire a lock named '{0}'.", name);
return false;
}
private DistributedLock AcquireLock(string name, TimeSpan maxValidFor, TimeSpan? timeout, string machineName, int? threadId) {
var @lock = AcquireLockInternal(name, maxValidFor, machineName, threadId, timeout);
if (@lock != null)
return @lock;
throw new TimeoutException("Could not acquire a lock within the specified amount of time.");
}
private DistributedLock AcquireLockInternal(string name, TimeSpan maxValidFor, string machineName, int? threadId, TimeSpan? timeout = null) {
try {
DistributedLockRecord record = null;
var acquired = Poll(() => (record = AcquireLockRecord(name, maxValidFor, machineName, threadId)) != null, timeout);
if (acquired) {
Logger.Debug("Successfully acquired a lock named '{0}'.", name);
return threadId != null
? DistributedLock.ForThread(this, name, machineName, threadId.Value, record.Id.ToString())
: DistributedLock.ForMachine(this, name, machineName, record.Id.ToString());
}
} }
catch (Exception ex) { catch (Exception ex) {
Logger.Error(ex, "An error occurred while trying to acquire a lock."); Logger.Error(ex, "Error while trying to acquire a lock named '{0}'.", name);
throw; throw;
} }
finally {
childLifetimeScope.Dispose(); return null;
}
private DistributedLockRecord AcquireLockRecord(string name, TimeSpan maxValidFor, string machineName, int? threadId) {
lock (_transactionManagerLock) {
var childLifetimeScope = CreateChildLifetimeScope(name);
try {
var transactionManager = childLifetimeScope.Resolve<ITransactionManager>();
transactionManager.RequireNew(IsolationLevel.ReadCommitted);
// This way we can create a nested transaction scope instead of having the unwanted effect
// of manipulating the transaction of the caller.
var repository = childLifetimeScope.Resolve<IRepository<DistributedLockRecord>>();
// Find an existing, active lock, if any.
var record = repository.Table.FirstOrDefault(x => x.Name == name && x.ValidUntilUtc >= _clock.UtcNow && x.Count > 0);
// The current owner name (based on machine name and current thread ID).
var canAcquireLock = false;
// Check if there's already an active lock.
if (record != null) {
// Check if the machine name assigned to the lock is the one trying to acquire it.
if (record.MachineName == machineName) {
if(record.ThreadId != threadId)
throw new InvalidOperationException(
threadId == null
? "An attempt to acquire a lock for a machine was detected while the requested lock is already assigned to a specific thread."
: "An attempt to acquire a lock for a thread was detected while the requested lock is already assigned to a machine.");
record.Count++;
canAcquireLock = true;
}
}
else {
// No one has an active lock yet, so good to go.
record = new DistributedLockRecord {
Name = name,
MachineName = machineName,
ThreadId = threadId,
Count = 1,
CreatedUtc = _clock.UtcNow,
ValidUntilUtc = _clock.UtcNow + maxValidFor
};
repository.Create(record);
canAcquireLock = true;
}
if (!canAcquireLock)
return null;
return record;
}
catch (Exception ex) {
Logger.Error(ex, "An error occurred while trying to acquire a lock.");
throw;
}
finally {
childLifetimeScope.Dispose();
}
} }
} }
private string GetOwnerName() { /// <summary>
return String.Format("{0}_{1}", _machineNameProvider.GetMachineName(), Thread.CurrentThread.ManagedThreadId); /// Executes the specified function until it returns true, for the specified amount of time, or indefinitely if no timeout was given.
/// </summary>
/// <param name="pollFunc">The function to repeatedly execute until it returns true.</param>
/// <param name="timeout">The amount of time to retry executing the function. If null is specified, the specified function is executed indefinitely until it returns true.</param>
/// <returns>Returns true if the specified function returned true within the specified timeout, false otherwise.</returns>
private bool Poll(Func<bool> pollFunc, TimeSpan? timeout) {
var waitedTime = TimeSpan.Zero;
var waitTime = TimeSpan.FromMilliseconds(timeout.GetValueOrDefault().TotalMilliseconds / 10);
bool acquired;
while (!(acquired = pollFunc()) && (timeout == null || waitedTime < timeout.Value)) {
Task.Delay(waitTime).ContinueWith(t => {
waitedTime += waitTime;
}).Wait();
}
return acquired;
}
private string GetMachineName() {
return _machineNameProvider.GetMachineName();
}
private int GetThreadId() {
return _threadProvider.GetCurrentThreadId();
} }
private ILifetimeScope CreateChildLifetimeScope(string name) { private ILifetimeScope CreateChildLifetimeScope(string name) {

View File

@@ -1,11 +0,0 @@
using System;
namespace Orchard.Tasks.Locking.Services {
/// <summary>
/// Represents a distributed lock.
/// </summary>
public interface IDistributedLock : IDisposable {
int Id { get; }
string Name { get; }
}
}

View File

@@ -6,27 +6,94 @@ namespace Orchard.Tasks.Locking.Services {
/// </summary> /// </summary>
public interface IDistributedLockService : ISingletonDependency { public interface IDistributedLockService : ISingletonDependency {
/// <summary> /// <summary>
/// Tries to acquire a lock on the specified name. /// Tries to acquire a lock on the specified name for the current machine.
/// </summary> /// </summary>
/// <param name="name">The name to use for the lock.</param> /// <param name="name">The name to use for the lock.</param>
/// <param name="maxLifetime">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param> /// <param name="maxValidFor">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <param name="timeout">The amount of time to wait for the lock to be acquired before timing out.</param> /// <param name="timeout">The amount of time to wait for the lock to be acquired before timing out. A null value will cause the method to return immedieately if no lock could be acquired.</param>
/// <param name="lock">The acquired lock.</param> /// <param name="lock">The acquired lock.</param>
/// <returns>Returns true if a lock was successfully acquired, false otherwise.</returns> /// <returns>Returns true if a lock was successfully acquired, false otherwise.</returns>
bool TryAcquireLock(string name, TimeSpan maxLifetime, TimeSpan timeout, out IDistributedLock @lock); bool TryAcquireLockForMachine(string name, TimeSpan maxValidFor, TimeSpan? timeout, out DistributedLock @lock);
/// <summary> /// <summary>
/// Acquires a lock with the specified parameters. /// Acquires a lock with the specified parameters for the current machine.
/// </summary> /// </summary>
/// <param name="name">The name to use for the lock.</param> /// <param name="name">The name to use for the lock.</param>
/// <param name="maxLifetime">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param> /// <param name="maxLifetime">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <param name="timeout">The amount of time to wait for the lock to be acquired before timing out.</param> /// <param name="timeout">The amount of time to wait for the lock to be acquired before timing out. A null value will cause the method to block indefinitely until a lock can be acquired.</param>
/// <returns>Returns a lock if one was successfully acquired, null otherwise.</returns> /// <returns>Returns a lock.</returns>
IDistributedLock AcquireLock(string name, TimeSpan maxLifetime, TimeSpan timeout); /// <exception cref="TimeoutException">Throws a TimeoutException if no lock could be acquired in time.</exception>
DistributedLock AcquireLockForMachine(string name, TimeSpan maxLifetime, TimeSpan? timeout);
/// <summary>
/// Tries to acquire a lock on the specified name for the current thread.
/// </summary>
/// <param name="name">The name to use for the lock.</param>
/// <param name="maxValidFor">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <param name="timeout">The amount of time to wait for the lock to be acquired before timing out. A null value will cause the method to return immedieately if no lock could be acquired.</param>
/// <param name="lock">The acquired lock.</param>
/// <returns>Returns true if a lock was successfully acquired, false otherwise.</returns>
bool TryAcquireLockForThread(string name, TimeSpan maxValidFor, TimeSpan? timeout, out DistributedLock @lock);
/// <summary>
/// Acquires a lock with the specified parameters for the current thread.
/// </summary>
/// <param name="name">The name to use for the lock.</param>
/// <param name="maxLifetime">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <param name="timeout">The amount of time to wait for the lock to be acquired before timing out. A null value will cause the method to block indefinitely until a lock can be acquired.</param>
/// <returns>Returns a lock.</returns>
/// <exception cref="TimeoutException">Throws a TimeoutException if no lock could be acquired in time.</exception>
DistributedLock AcquireLockForThread(string name, TimeSpan maxLifetime, TimeSpan? timeout);
/// <summary> /// <summary>
/// Disposes the specified lock. /// Disposes the specified lock.
/// </summary> /// </summary>
void DisposeLock(IDistributedLock @lock); void ReleaseLock(DistributedLock @lock);
}
public static class DistributedLockServiceExtensions {
/// <summary>
/// Tries to acquire a lock on the specified name for the current machine.
/// </summary>
/// <param name="name">The name to use for the lock.</param>
/// <param name="maxValidFor">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <param name="lock">The acquired lock.</param>
/// <returns>Returns true if a lock was successfully acquired, false otherwise.</returns>
public static bool TryAcquireLockForMachine(this IDistributedLockService service, string name, TimeSpan maxValidFor, out DistributedLock @lock) {
return service.TryAcquireLockForMachine(name, maxValidFor, null, out @lock);
}
/// <summary>
/// Acquires a lock with the specified parameters for the current machine.
/// </summary>
/// <param name="name">The name to use for the lock.</param>
/// <param name="maxLifetime">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <returns>Returns a lock.</returns>
/// <exception cref="TimeoutException">Throws a TimeoutException if no lock could be acquired in time.</exception>
public static DistributedLock AcquireLockForMachine(this IDistributedLockService service, string name, TimeSpan maxLifetime) {
return service.AcquireLockForMachine(name, maxLifetime, null);
}
/// <summary>
/// Tries to acquire a lock on the specified name for the current thread.
/// </summary>
/// <param name="name">The name to use for the lock.</param>
/// <param name="maxValidFor">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <param name="lock">The acquired lock.</param>
/// <returns>Returns true if a lock was successfully acquired, false otherwise.</returns>
public static bool TryAcquireLockForThread(this IDistributedLockService service, string name, TimeSpan maxValidFor, out DistributedLock @lock) {
return service.TryAcquireLockForThread(name, maxValidFor, null, out @lock);
}
/// <summary>
/// Acquires a lock with the specified parameters for the current thread.
/// </summary>
/// <param name="name">The name to use for the lock.</param>
/// <param name="maxLifetime">The maximum amount of time the lock is allowed. This is a safety net in case the caller fails to release the lock.</param>
/// <returns>Returns a lock.</returns>
/// <exception cref="TimeoutException">Throws a TimeoutException if no lock could be acquired in time.</exception>
public static DistributedLock AcquireLockForThread(this IDistributedLockService service, string name, TimeSpan maxLifetime) {
return service.AcquireLockForThread(name, maxLifetime, null);
}
} }
} }

View File

@@ -1,31 +0,0 @@
using Orchard.Logging;
namespace Orchard.Tasks.Locking.Services {
/// <summary>
/// Provides a database driven implementation of <see cref="IDistributedLock" />
/// </summary>
public class Lock : IDistributedLock {
private readonly IDistributedLockService _distributedLockService;
public string Name { get; set; }
private bool _isDisposed;
public Lock(IDistributedLockService distributedLockService, string name, int id) {
_distributedLockService = distributedLockService;
Name = name;
Id = id;
}
public ILogger Logger { get; set; }
public int Id { get; set; }
// This will be called at least and at the latest by the IoC container when the request ends.
public void Dispose() {
if (!_isDisposed) {
_isDisposed = true;
_distributedLockService.DisposeLock(this);
}
}
}
}