Skip to content

Commit 949e904

Browse files
odinserjodinserj
committed
Fix parallelism-related tests for delayed and recurring job schedulers
1 parent 018d1e5 commit 949e904

2 files changed

Lines changed: 69 additions & 30 deletions

File tree

‎tests/Hangfire.Core.Tests/Server/DelayedJobSchedulerFacts.cs‎

Lines changed: 45 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -42,22 +42,31 @@ public DelayedJobSchedulerFacts()
4242

4343
_connection
4444
.Setup(x => x.GetFirstByLowestScoreFromSet("schedule", 0, It.Is<double>(time => time > 0)))
45-
.Returns(_schedule.FirstOrDefault);
45+
.Returns(() =>
46+
{
47+
lock (_schedule) return _schedule.FirstOrDefault();
48+
});
4649

4750
_connection
4851
.Setup(x => x.GetFirstByLowestScoreFromSet("schedule", 0, It.Is<double>(time => time > 0), It.IsAny<int>()))
49-
.Returns(_schedule.ToList);
52+
.Returns(() =>
53+
{
54+
lock (_schedule) return _schedule.ToList();
55+
});
5056

5157
_stateChanger
5258
.Setup(x => x.ChangeState(It.IsNotNull<StateChangeContext>()))
5359
.Callback<StateChangeContext>(ctx =>
5460
{
55-
if (!(ctx.NewState is ScheduledState)) _schedule.Remove(ctx.BackgroundJobId);
61+
if (!(ctx.NewState is ScheduledState)) lock (_schedule) _schedule.Remove(ctx.BackgroundJobId);
5662
});
5763

5864
_transaction
5965
.Setup(x => x.RemoveFromSet("schedule", It.IsNotNull<string>()))
60-
.Callback<string, string>((key, value) => _schedule.Remove(value));
66+
.Callback<string, string>((key, value) =>
67+
{
68+
lock (_schedule) _schedule.Remove(value);
69+
});
6170
}
6271

6372
[Fact]
@@ -79,7 +88,7 @@ public void Execute_MovesJobStateToEnqueued(bool batching, int maxParallelism)
7988
EnableBatching(batching);
8089

8190
var scheduler = CreateScheduler(maxParallelism);
82-
_schedule.Add(JobId);
91+
lock (_schedule) _schedule.Add(JobId);
8392

8493
scheduler.Execute(_context.Object);
8594

@@ -102,8 +111,11 @@ public void Execute_EnqueuesJobIdDirectly_AndRemovesItFromSchedule_WhenTargetQue
102111
EnableBatching(batching);
103112

104113
// Arrange
105-
_schedule.Add("default:some-id");
106-
_schedule.Add("critical:another-id");
114+
lock (_schedule)
115+
{
116+
_schedule.Add("default:some-id");
117+
_schedule.Add("critical:another-id");
118+
}
107119

108120
var scheduler = CreateScheduler(maxParallelism);
109121

@@ -128,8 +140,11 @@ public void Execute_MovesJobStateToEnqueued_UsingBatching_WhenAvailable(int maxP
128140
// Arrange
129141
EnableBatching(true);
130142

131-
_schedule.Add("job-1");
132-
_schedule.Add("job-2");
143+
lock (_schedule)
144+
{
145+
_schedule.Add("job-1");
146+
_schedule.Add("job-2");
147+
}
133148

134149
var scheduler = CreateScheduler(maxParallelism);
135150

@@ -153,8 +168,11 @@ public void Execute_WithBatching_EnqueuesJobIdDirectly_AndRemovesItFromSchedule_
153168
// Arrange
154169
EnableBatching(true);
155170

156-
_schedule.Add("default:some-id");
157-
_schedule.Add("critical:another-id");
171+
lock (_schedule)
172+
{
173+
_schedule.Add("default:some-id");
174+
_schedule.Add("critical:another-id");
175+
}
158176

159177
var scheduler = CreateScheduler(maxParallelism);
160178

@@ -181,11 +199,14 @@ public void Execute_DoesNotUseBatching_WhenConnectionMethod_ThrowsAnException(in
181199
.Setup(x => x.GetFirstByLowestScoreFromSet(It.IsAny<string>(), It.IsAny<double>(), It.IsAny<double>(),It.IsAny<int>()))
182200
.Throws<NotImplementedException>();
183201

184-
_schedule.Add("job-1");
202+
lock (_schedule) _schedule.Add("job-1");
185203

186204
_connection
187205
.Setup(x => x.GetFirstByLowestScoreFromSet("schedule", 0, It.Is<double>(time => time > 0)))
188-
.Returns(_schedule.FirstOrDefault);
206+
.Returns(() =>
207+
{
208+
lock (_schedule) return _schedule.FirstOrDefault();
209+
});
189210

190211
var scheduler = CreateScheduler(maxParallelism);
191212

@@ -226,7 +247,7 @@ public void Execute_RemovesAJobIdentifierFromTheSet_WhenStateChangeFails(bool ba
226247
_stateChanger
227248
.Setup(x => x.ChangeState(It.IsAny<StateChangeContext>()))
228249
.Returns<IState>(null);
229-
_schedule.Add(JobId);
250+
lock (_schedule) _schedule.Add(JobId);
230251

231252
var scheduler = CreateScheduler(maxParallelism);
232253

@@ -246,7 +267,7 @@ public void Execute_MovesJobToTheFailedState_WithFiltersDisabled_WhenStateChange
246267
// Arrange
247268
EnableBatching(batching);
248269

249-
_schedule.Add(JobId);
270+
lock (_schedule) _schedule.Add(JobId);
250271
_stateChanger
251272
.Setup(x => x.ChangeState(It.Is<StateChangeContext>(ctx => ctx.NewState is EnqueuedState)))
252273
.Throws<InvalidOperationException>();
@@ -282,8 +303,11 @@ public void Execute_AbleToProcessFurtherJobs_WhenStateChangerThrowsAnException_F
282303
// Arrange
283304
EnableBatching(batching);
284305

285-
_schedule.Add(JobId);
286-
_schedule.Add("AnotherId");
306+
lock (_schedule)
307+
{
308+
_schedule.Add(JobId);
309+
_schedule.Add("AnotherId");
310+
}
287311

288312
_stateChanger
289313
.Setup(x => x.ChangeState(It.Is<StateChangeContext>(ctx => ctx.BackgroundJobId == JobId && ctx.NewState is ScheduledState)))
@@ -342,7 +366,7 @@ public void Execute_RemovesJobFromSchedule_WhenIdDoesNotExists(bool batching, in
342366
{
343367
// Arrange
344368
EnableBatching(batching);
345-
_schedule.Add(JobId);
369+
lock (_schedule) _schedule.Add(JobId);
346370

347371
_connection.Setup(x => x.GetJobData(JobId)).Returns<JobData>(null);
348372

@@ -370,7 +394,7 @@ public void Execute_RemovesJobFromSchedule_WhenJobIsNotInScheduledState(bool bat
370394
{
371395
// Arrange
372396
EnableBatching(batching);
373-
_schedule.Add(JobId);
397+
lock (_schedule) _schedule.Add(JobId);
374398

375399
_connection.Setup(x => x.GetJobData(JobId))
376400
.Returns(new JobData { State = SucceededState.StateName });
@@ -399,15 +423,15 @@ public void Execute_DoesNotRemoveJobFromSchedule_WhenJobIsInTheScheduledState(bo
399423
{
400424
// Arrange
401425
EnableBatching(batching);
402-
_schedule.Add(JobId);
426+
lock (_schedule) _schedule.Add(JobId);
403427

404428
_connection.Setup(x => x.GetJobData(JobId))
405429
.Returns(new JobData { State = ScheduledState.StateName });
406430

407431
_stateChanger
408432
.SetupSequence(x => x.ChangeState(It.Is<StateChangeContext>(ctx => ctx.NewState is EnqueuedState)))
409433
.Returns((IState)null)
410-
.Returns(() => { _schedule.Remove(JobId); return new EnqueuedState(); });
434+
.Returns(() => { lock (_schedule) _schedule.Remove(JobId); return new EnqueuedState(); });
411435

412436
var scheduler = CreateScheduler(maxParallelism);
413437

‎tests/Hangfire.Core.Tests/Server/RecurringJobSchedulerFacts.cs‎

Lines changed: 24 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -66,21 +66,33 @@ public RecurringJobSchedulerFacts()
6666
_context.Storage.Setup(x => x.GetConnection()).Returns(_connection.Object);
6767

6868
_connection.Setup(x => x.GetFirstByLowestScoreFromSet("recurring-jobs", 0, JobHelper.ToTimestamp(_nowInstant)))
69-
.Returns(() => _schedule.FirstOrDefault());
69+
.Returns(() =>
70+
{
71+
lock (_schedule) return _schedule.FirstOrDefault();
72+
});
7073

7174
_connection.Setup(x => x.GetAllEntriesFromHash($"recurring-job:{RecurringJobId}"))
7275
.Returns(_recurringJob);
7376

7477
_connection.SetupSequence(x => x.GetFirstByLowestScoreFromSet("recurring-jobs", 0, JobHelper.ToTimestamp(_nowInstant), It.IsAny<int>()))
75-
.Returns(() => _schedule.ToList());
78+
.Returns(() =>
79+
{
80+
lock (_schedule) return _schedule.ToList();
81+
});
7682

7783
_transaction = new Mock<IWriteOnlyTransaction>();
7884
_transaction
7985
.Setup(x => x.RemoveFromSet("recurring-jobs", It.IsNotNull<string>()))
80-
.Callback<string, string>((key, value) => _schedule.Remove(value));
86+
.Callback<string, string>((key, value) =>
87+
{
88+
lock (_schedule) _schedule.Remove(value);
89+
});
8190
_transaction
8291
.Setup(x => x.AddToSet("recurring-jobs", It.IsNotNull<string>(), It.IsAny<double>()))
83-
.Callback<string, string, double>((key, value, score) => _schedule.Remove(value));
92+
.Callback<string, string, double>((key, value, score) =>
93+
{
94+
lock (_schedule) _schedule.Remove(value);
95+
});
8496

8597
_connection.Setup(x => x.CreateWriteTransaction()).Returns(_transaction.Object);
8698

@@ -362,9 +374,12 @@ public void Execute_RemovesRecurringJobFromSchedule_WhenHashDoesNotExist(bool ba
362374
{
363375
// Arrange
364376
EnableBatching(batching);
365-
366-
_schedule.Clear();
367-
_schedule.Add("non-existing-job");
377+
378+
lock (_schedule)
379+
{
380+
_schedule.Clear();
381+
_schedule.Add("non-existing-job");
382+
}
368383

369384
var scheduler = CreateScheduler(maxParallelism);
370385

@@ -1353,7 +1368,7 @@ public void Execute_AbleToProcessFurtherJobs_WhenStateChangerThrowsAnException_F
13531368
{
13541369
// Arrange
13551370
EnableBatching(batching);
1356-
_schedule.Add("AnotherId");
1371+
lock (_schedule) _schedule.Add("AnotherId");
13571372
_connection.Setup(x => x.GetAllEntriesFromHash("recurring-job:AnotherId"))
13581373
.Returns(_recurringJob);
13591374

@@ -1381,7 +1396,7 @@ public void Execute_RemovesNonExistingRecurringJobFromSet_AndDoesNotStopPipeline
13811396
{
13821397
// Arrange
13831398
EnableBatching(batching);
1384-
_schedule.Add("AnotherId");
1399+
lock (_schedule) _schedule.Add("AnotherId");
13851400
_connection.Setup(x => x.GetAllEntriesFromHash("recurring-job:AnotherId")).Returns(_recurringJob);
13861401
_connection.Setup(x => x.GetAllEntriesFromHash($"recurring-job:{RecurringJobId}")).Returns<Dictionary<string, string>>(null);
13871402

0 commit comments

Comments
 (0)