到目前为止,我已经避免了测试多线程代码的噩梦,因为它似乎是一个太大的雷区。我想知道人们是如何测试依赖于线程的代码以获得成功执行的,或者人们是如何测试那些仅在两个线程以给定方式交互时才会出现的问题的?
对于今天的程序员来说,这似乎是一个非常关键的问题,恕我直言,将我们的知识集中在这个问题上是很有用的。
到目前为止,我已经避免了测试多线程代码的噩梦,因为它似乎是一个太大的雷区。我想知道人们是如何测试依赖于线程的代码以获得成功执行的,或者人们是如何测试那些仅在两个线程以给定方式交互时才会出现的问题的?
对于今天的程序员来说,这似乎是一个非常关键的问题,恕我直言,将我们的知识集中在这个问题上是很有用的。
当前回答
有一些很好的工具。下面是一些Java的摘要。
一些好的静态分析工具包括FindBugs(提供了一些有用的提示)、JLint、Java Pathfinder (JPF & JPF2)和Bogor。
multithreaddtc是一个非常好的动态分析工具(集成到JUnit中),您必须在其中设置自己的测试用例。
IBM研究院的竞赛很有趣。它通过插入各种线程修改行为(例如sleep & yield)来检测你的代码,试图随机发现错误。
SPIN是对Java(和其他)组件建模的一个非常酷的工具,但是您需要一些有用的框架。它很难使用,但如果你知道如何使用它,它是非常强大的。相当多的工具在底层使用SPIN。
multithreaddtc可能是最主流的,但是上面列出的一些静态分析工具绝对值得一看。
其他回答
它并不完美,但我用c#写了这个帮助程序:
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
namespace Proto.Promises.Tests.Threading
{
public class ThreadHelper
{
public static readonly int multiThreadCount = Environment.ProcessorCount * 100;
private static readonly int[] offsets = new int[] { 0, 10, 100, 1000 };
private readonly Stack<Task> _executingTasks = new Stack<Task>(multiThreadCount);
private readonly Barrier _barrier = new Barrier(1);
private int _currentParticipants = 0;
private readonly TimeSpan _timeout;
public ThreadHelper() : this(TimeSpan.FromSeconds(10)) { } // 10 second timeout should be enough for most cases.
public ThreadHelper(TimeSpan timeout)
{
_timeout = timeout;
}
/// <summary>
/// Execute the action multiple times in parallel threads.
/// </summary>
public void ExecuteMultiActionParallel(Action action)
{
for (int i = 0; i < multiThreadCount; ++i)
{
AddParallelAction(action);
}
ExecutePendingParallelActions();
}
/// <summary>
/// Execute the action once in a separate thread.
/// </summary>
public void ExecuteSingleAction(Action action)
{
AddParallelAction(action);
ExecutePendingParallelActions();
}
/// <summary>
/// Add an action to be run in parallel.
/// </summary>
public void AddParallelAction(Action action)
{
var taskSource = new TaskCompletionSource<bool>();
lock (_executingTasks)
{
++_currentParticipants;
_barrier.AddParticipant();
_executingTasks.Push(taskSource.Task);
}
new Thread(() =>
{
try
{
_barrier.SignalAndWait(); // Try to make actions run in lock-step to increase likelihood of breaking race conditions.
action.Invoke();
taskSource.SetResult(true);
}
catch (Exception e)
{
taskSource.SetException(e);
}
}).Start();
}
/// <summary>
/// Runs the pending actions in parallel, attempting to run them in lock-step.
/// </summary>
public void ExecutePendingParallelActions()
{
Task[] tasks;
lock (_executingTasks)
{
_barrier.SignalAndWait();
_barrier.RemoveParticipants(_currentParticipants);
_currentParticipants = 0;
tasks = _executingTasks.ToArray();
_executingTasks.Clear();
}
try
{
if (!Task.WaitAll(tasks, _timeout))
{
throw new TimeoutException($"Action(s) timed out after {_timeout}, there may be a deadlock.");
}
}
catch (AggregateException e)
{
// Only throw one exception instead of aggregate to try to avoid overloading the test error output.
throw e.Flatten().InnerException;
}
}
/// <summary>
/// Run each action in parallel multiple times with differing offsets for each run.
/// <para/>The number of runs is 4^actions.Length, so be careful if you don't want the test to run too long.
/// </summary>
/// <param name="expandToProcessorCount">If true, copies each action on additional threads up to the processor count. This can help test more without increasing the time it takes to complete.
/// <para/>Example: 2 actions with 6 processors, runs each action 3 times in parallel.</param>
/// <param name="setup">The action to run before each parallel run.</param>
/// <param name="teardown">The action to run after each parallel run.</param>
/// <param name="actions">The actions to run in parallel.</param>
public void ExecuteParallelActionsWithOffsets(bool expandToProcessorCount, Action setup, Action teardown, params Action[] actions)
{
setup += () => { };
teardown += () => { };
int actionCount = actions.Length;
int expandCount = expandToProcessorCount ? Math.Max(Environment.ProcessorCount / actionCount, 1) : 1;
foreach (var combo in GenerateCombinations(offsets, actionCount))
{
setup.Invoke();
for (int k = 0; k < expandCount; ++k)
{
for (int i = 0; i < actionCount; ++i)
{
int offset = combo[i];
Action action = actions[i];
AddParallelAction(() =>
{
for (int j = offset; j > 0; --j) { } // Just spin in a loop for the offset.
action.Invoke();
});
}
}
ExecutePendingParallelActions();
teardown.Invoke();
}
}
// Input: [1, 2, 3], 3
// Ouput: [
// [1, 1, 1],
// [2, 1, 1],
// [3, 1, 1],
// [1, 2, 1],
// [2, 2, 1],
// [3, 2, 1],
// [1, 3, 1],
// [2, 3, 1],
// [3, 3, 1],
// [1, 1, 2],
// [2, 1, 2],
// [3, 1, 2],
// [1, 2, 2],
// [2, 2, 2],
// [3, 2, 2],
// [1, 3, 2],
// [2, 3, 2],
// [3, 3, 2],
// [1, 1, 3],
// [2, 1, 3],
// [3, 1, 3],
// [1, 2, 3],
// [2, 2, 3],
// [3, 2, 3],
// [1, 3, 3],
// [2, 3, 3],
// [3, 3, 3]
// ]
private static IEnumerable<int[]> GenerateCombinations(int[] options, int count)
{
int[] indexTracker = new int[count];
int[] combo = new int[count];
for (int i = 0; i < count; ++i)
{
combo[i] = options[0];
}
// Same algorithm as picking a combination lock.
int rollovers = 0;
while (rollovers < count)
{
yield return combo; // No need to duplicate the array since we're just reading it.
for (int i = 0; i < count; ++i)
{
int index = ++indexTracker[i];
if (index == options.Length)
{
indexTracker[i] = 0;
combo[i] = options[0];
if (i == rollovers)
{
++rollovers;
}
}
else
{
combo[i] = options[index];
break;
}
}
}
}
}
}
使用示例:
[Test]
public void DeferredMayBeBeResolvedAndPromiseAwaitedConcurrently_void0()
{
Promise.Deferred deferred = default(Promise.Deferred);
Promise promise = default(Promise);
int invokedCount = 0;
var threadHelper = new ThreadHelper();
threadHelper.ExecuteParallelActionsWithOffsets(false,
// Setup
() =>
{
invokedCount = 0;
deferred = Promise.NewDeferred();
promise = deferred.Promise;
},
// Teardown
() => Assert.AreEqual(1, invokedCount),
// Parallel Actions
() => deferred.Resolve(),
() => promise.Then(() => { Interlocked.Increment(ref invokedCount); }).Forget()
);
}
一个简单的测试模式可以用于一些(不是所有!)用例,就是多次重复相同的测试。例如,假设你有一个方法:
def process(input):
# Spawns several threads to do the job
# ...
return output
创建一堆测试:
process(input1) -> expect to return output1
process(input2) -> expect to return output2
...
现在将每个测试运行多次。
如果流程的实现包含一个微小的错误(例如死锁、竞态条件等),出现的概率为0.1%,那么运行1000次测试,则该错误至少出现一次的概率为64%。运行测试10000次,得到>99%的概率。
有一篇关于这个主题的文章,在示例代码中使用Rust作为语言:
https://medium.com/@polyglot_factotum/rust-concurrency-five-easy-pieces-871f1c62906a
总而言之,诀窍在于编写并发逻辑,使其对涉及多个执行线程的非确定性具有健壮性,使用通道和condvars等工具。
然后,如果这就是您构建“组件”的方式,那么测试它们的最简单方法是使用通道向它们发送消息,然后阻塞其他通道以断言组件发送某些预期的消息。
链接到的文章完全使用单元测试编写。
Pete Goodliffe有一个关于线程代码单元测试的系列。
是很困难的。我采用了更简单的方法,尽量将线程代码从实际测试中抽象出来。皮特确实提到了我分手的方式是错误的但我要么是正确的,要么就是我很幸运。
(如果可能的话)不要使用线程,使用actor /活动对象。易于测试。