Event.BeginInvoke + Quartz.NET + FileSystemWatcher + Task.Run 循环 = 异常？
Event.BeginInvoke + Quartz.NET + FileSystemWatcher + Task.Run in loop = exception?
我查看了 SO，但找不到与我相当独特的问题相匹配的内容。
问题：
我一直在使用 Quartz.NET 和 FileSystemWatcher 编写一些代码，基本设置如下。
Quartz.NET 以 15 秒的间隔触发计时器。此触发器导致在特定目录中创建文件，文件名格式为 *.csv（或 *.xml——仅作为示例）。 FileSystemWatcher 注意到该文件并执行一些代码，在 return 中在同一目录中吐出一个 *.csv。现在我知道这将重新触发观察者并且它将陷入无限循环——我这一次打算这样做。
运行此代码时，我在管理事件的处理程序代码中收到越界异常。值得注意的是，这个异常是在处理 Quartz 作业和使用 for 循环时引起的。异常不会发生在foreach
我意识到这段代码可能...很有趣，但是任何人都可以破译什么会导致 for 中的额外增量（范围超出范围异常）以及为什么迭代器会成功？
2016 年 4 月 14 日更新 8:05AM 美国东部时间
正如乔恩指出的那样，这是一个简单的疏忽和睡眠不足。任务执行时i是共享资源.
var tl = new Task[pdList.Count];
for (int i = 0; i < pdList.Count; i++)
{
int li = i; // i is shared...use local copy
// use foreach instead
tl[i] = Task.Factory.StartNew(() => { ExecuteProcess(pdList[li], hargs); }, TaskCreationOptions.LongRunning);
}
2016 年 4 月 14 日更新 7:09AM 美国东部时间
根据 Jon 的指导，我将编写一小段代码，有望重现该问题。与此同时，我添加了堆栈跟踪和异常详细信息。整个项目可在 github 上免费获得：
https://github.com/banjoCoder/ThatIntegrationEngine/tree/Dev
异常信息
Index was out of range. Must be non-negative and less than the size of
the collection. Parameter name: index
异常堆栈跟踪：
at System.ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument argument, ExceptionResource resource)
at System.Collections.Generic.List`1.get_Item(Int32 index)
at ThatIntegrationEngine.Engine.<>c__DisplayClass25_1.<BeginProcessExecution>b__0() in Z:\src\ThatIntegrationEngine\ThatIntegrationEngine.Core\Hood\Engine.cs:line 265
at System.Threading.Tasks.Task.InnerInvoke()
at System.Threading.Tasks.Task.Execute()
任务线程内的堆栈跟踪调用：
at System.Environment.GetStackTrace(Exception e, Boolean needFileInfo)
at System.Environment.get_StackTrace()
at ThatIntegrationEngine.Engine.<>c__DisplayClass25_1.<BeginProcessExecution>b__0() in Z:\src\ThatIntegrationEngine\ThatIntegrationEngine.Core\Hood\Engine.cs:line 263
at System.Threading.Tasks.Task.InnerInvoke()
at System.Threading.Tasks.Task.Execute()
at System.Threading.Tasks.Task.ExecutionContextCallback(Object obj)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot)
at System.Threading.Tasks.Task.ExecuteEntry(Boolean bPreventDoubleExecution)
at System.Threading.Tasks.ThreadPoolTaskScheduler.LongRunningThreadWork(Object obj)
at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.ThreadHelper.ThreadStart(Object obj)
at System.Environment.GetStackTrace(Exception e, Boolean needFileInfo)
at System.Environment.get_StackTrace()
at ThatIntegrationEngine.Engine.<>c__DisplayClass25_1.<BeginProcessExecution>b__0() in Z:\src\ThatIntegrationEngine\ThatIntegrationEngine.Core\Hood\Engine.cs:line 263
at System.Threading.Tasks.Task.InnerInvoke()
at System.Threading.Tasks.Task.Execute()
at System.Threading.Tasks.Task.ExecutionContextCallback(Object obj)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot)
at System.Threading.Tasks.Task.ExecuteEntry(Boolean bPreventDoubleExecution)
at System.Threading.Tasks.ThreadPoolTaskScheduler.LongRunningThreadWork(Object obj)
at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.ThreadHelper.ThreadStart(Object obj)
Exception thrown: 'System.ArgumentOutOfRangeException' in mscorlib.dll
代码流：
Quartz 触发器触发时引发的事件：
protected virtual void OnTimedActionAsync(object s, CronSchedulerEventArgs e)
{
EventHandler<CronSchedulerEventArgs> handler = TimedActionAsync;
handler?.BeginInvoke(this, e, cs =>
{
var delg = (EventHandler<CronSchedulerEventArgs>)cs.AsyncState;
delg.EndInvoke(cs);
},
handler);
}
内部引发的异步事件 class 订阅了 FSW 创建的事件
protected virtual void OnCreatedAsync(object sender, FileSystemEventArgs args)
{
EventHandler<DirWatcherEventArgs> handler = FileArrivedAsync;
handler?.BeginInvoke(this, new DirWatcherEventArgs(args.FullPath, WatcherChangeTypes.Created, Id, Name)
, cs =>
{
var delg = (EventHandler<DirWatcherEventArgs>)cs.AsyncState;
delg.EndInvoke(cs);
}
, handler);
}
事件终于处理完毕
private void HandleScheduledActionAsync(object sender, CronSchedulerEventArgs csarg)
{
var scheduler = sender as ICronScheduler;
if (scheduler != null)
{
// load process type based on trigger (sender) id
// dynamically create instance of the process and pass
// all expected information
BeginProcessExecution(csarg);
}
}
private void HandleFileArrivalAsync(object sender, DirWatcherEventArgs dwargs)
{
var watcher = sender as IDirectoryWatcher;
if (watcher != null)
{
// load process type based on trigger (sender) id
// dynamically create instance of the process and pass
// all expected information
BeginProcessExecution(dwargs);
}
}
private void BeginProcessExecution(HandlerEventArgs hargs)
{
IList<IProcessDetails> pdList = null;
if (this._processHandlerRelation.TryGetValue(hargs.TriggerId, out pdList))
{
if (pdList.Count > 1)
{
try
{
//!!! FAILURE !!!
//this fails with out of range exception
var tl = new Task[pdList.Count];
for(int i = 0; i < pdList.Count; i++)
{
//!!! EXCEPTION !!!
// this causes an out of range exception
// at exec task is attempting to read final value of i
// is i accessed by ref? -- no it's a shared resource
tl[i] = Task.Factory.StartNew(() => ExecuteProcess(pdList[i], hargs), TaskCreationOptions.LongRunning);
}
//this does not cause an exception
//var tl = new List<Task>();
//foreach (var pd in pdList)
//{
// tl.Add(Task.Factory.StartNew(() => ExecuteProcess(pd, hargs), TaskCreationOptions.LongRunning));
//}
Task.WaitAll(tl);
// this does not cause an exception
//Parallel.ForEach(pdList, pd => ExecuteProcess(pd, hargs));
}
catch (Exception e)
{
// log exception
}
}
else
{// avoid parallelization if only one process is expected
// to be executed
// execute on calling thread
ExecuteProcess(pdList[0], hargs);
}
}
}
正如 Jon Skeet 指出的那样，这是一个简单的疏忽和睡眠不足。当任务执行时，我是共享资源。
var tl = new Task[pdList.Count];
for (int i = 0; i < pdList.Count; i++)
{
int li = i; // i is shared...use local copy
// use foreach instead
tl[i] = Task.Factory.StartNew(() => ExecuteProcess(pdList[li], hargs), TaskCreationOptions.LongRunning);
}
工作解决方案：
int pi = 0;
var plist = new Task[pdList.Count];
foreach(var pd in pdList)
{
plist[pi++] = Task.Factory.StartNew(() => ExecuteProcess(pd, hargs), TaskCreationOptions.LongRunning);
}
Task.WaitAll(plist);
我查看了 SO，但找不到与我相当独特的问题相匹配的内容。
问题： 我一直在使用 Quartz.NET 和 FileSystemWatcher 编写一些代码，基本设置如下。
Quartz.NET 以 15 秒的间隔触发计时器。此触发器导致在特定目录中创建文件，文件名格式为 *.csv（或 *.xml——仅作为示例）。 FileSystemWatcher 注意到该文件并执行一些代码，在 return 中在同一目录中吐出一个 *.csv。现在我知道这将重新触发观察者并且它将陷入无限循环——我这一次打算这样做。
运行此代码时，我在管理事件的处理程序代码中收到越界异常。值得注意的是，这个异常是在处理 Quartz 作业和使用 for 循环时引起的。异常不会发生在foreach
我意识到这段代码可能...很有趣，但是任何人都可以破译什么会导致 for 中的额外增量（范围超出范围异常）以及为什么迭代器会成功？
2016 年 4 月 14 日更新 8:05AM 美国东部时间
正如乔恩指出的那样，这是一个简单的疏忽和睡眠不足。任务执行时i是共享资源.
var tl = new Task[pdList.Count];
for (int i = 0; i < pdList.Count; i++)
{
int li = i; // i is shared...use local copy
// use foreach instead
tl[i] = Task.Factory.StartNew(() => { ExecuteProcess(pdList[li], hargs); }, TaskCreationOptions.LongRunning);
}
2016 年 4 月 14 日更新 7:09AM 美国东部时间
根据 Jon 的指导，我将编写一小段代码，有望重现该问题。与此同时，我添加了堆栈跟踪和异常详细信息。整个项目可在 github 上免费获得：
https://github.com/banjoCoder/ThatIntegrationEngine/tree/Dev
异常信息
Index was out of range. Must be non-negative and less than the size of the collection. Parameter name: index
异常堆栈跟踪：
at System.ThrowHelper.ThrowArgumentOutOfRangeException(ExceptionArgument argument, ExceptionResource resource)
at System.Collections.Generic.List`1.get_Item(Int32 index)
at ThatIntegrationEngine.Engine.<>c__DisplayClass25_1.<BeginProcessExecution>b__0() in Z:\src\ThatIntegrationEngine\ThatIntegrationEngine.Core\Hood\Engine.cs:line 265
at System.Threading.Tasks.Task.InnerInvoke()
at System.Threading.Tasks.Task.Execute()
任务线程内的堆栈跟踪调用：
at System.Environment.GetStackTrace(Exception e, Boolean needFileInfo)
at System.Environment.get_StackTrace()
at ThatIntegrationEngine.Engine.<>c__DisplayClass25_1.<BeginProcessExecution>b__0() in Z:\src\ThatIntegrationEngine\ThatIntegrationEngine.Core\Hood\Engine.cs:line 263
at System.Threading.Tasks.Task.InnerInvoke()
at System.Threading.Tasks.Task.Execute()
at System.Threading.Tasks.Task.ExecutionContextCallback(Object obj)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot)
at System.Threading.Tasks.Task.ExecuteEntry(Boolean bPreventDoubleExecution)
at System.Threading.Tasks.ThreadPoolTaskScheduler.LongRunningThreadWork(Object obj)
at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.ThreadHelper.ThreadStart(Object obj)
at System.Environment.GetStackTrace(Exception e, Boolean needFileInfo)
at System.Environment.get_StackTrace()
at ThatIntegrationEngine.Engine.<>c__DisplayClass25_1.<BeginProcessExecution>b__0() in Z:\src\ThatIntegrationEngine\ThatIntegrationEngine.Core\Hood\Engine.cs:line 263
at System.Threading.Tasks.Task.InnerInvoke()
at System.Threading.Tasks.Task.Execute()
at System.Threading.Tasks.Task.ExecutionContextCallback(Object obj)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.Tasks.Task.ExecuteWithThreadLocal(Task& currentTaskSlot)
at System.Threading.Tasks.Task.ExecuteEntry(Boolean bPreventDoubleExecution)
at System.Threading.Tasks.ThreadPoolTaskScheduler.LongRunningThreadWork(Object obj)
at System.Threading.ThreadHelper.ThreadStart_Context(Object state)
at System.Threading.ExecutionContext.RunInternal(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state, Boolean preserveSyncCtx)
at System.Threading.ExecutionContext.Run(ExecutionContext executionContext, ContextCallback callback, Object state)
at System.Threading.ThreadHelper.ThreadStart(Object obj)
Exception thrown: 'System.ArgumentOutOfRangeException' in mscorlib.dll
代码流：
Quartz 触发器触发时引发的事件：
protected virtual void OnTimedActionAsync(object s, CronSchedulerEventArgs e)
{
EventHandler<CronSchedulerEventArgs> handler = TimedActionAsync;
handler?.BeginInvoke(this, e, cs =>
{
var delg = (EventHandler<CronSchedulerEventArgs>)cs.AsyncState;
delg.EndInvoke(cs);
},
handler);
}
内部引发的异步事件 class 订阅了 FSW 创建的事件
protected virtual void OnCreatedAsync(object sender, FileSystemEventArgs args)
{
EventHandler<DirWatcherEventArgs> handler = FileArrivedAsync;
handler?.BeginInvoke(this, new DirWatcherEventArgs(args.FullPath, WatcherChangeTypes.Created, Id, Name)
, cs =>
{
var delg = (EventHandler<DirWatcherEventArgs>)cs.AsyncState;
delg.EndInvoke(cs);
}
, handler);
}
事件终于处理完毕
private void HandleScheduledActionAsync(object sender, CronSchedulerEventArgs csarg)
{
var scheduler = sender as ICronScheduler;
if (scheduler != null)
{
// load process type based on trigger (sender) id
// dynamically create instance of the process and pass
// all expected information
BeginProcessExecution(csarg);
}
}
private void HandleFileArrivalAsync(object sender, DirWatcherEventArgs dwargs)
{
var watcher = sender as IDirectoryWatcher;
if (watcher != null)
{
// load process type based on trigger (sender) id
// dynamically create instance of the process and pass
// all expected information
BeginProcessExecution(dwargs);
}
}
private void BeginProcessExecution(HandlerEventArgs hargs)
{
IList<IProcessDetails> pdList = null;
if (this._processHandlerRelation.TryGetValue(hargs.TriggerId, out pdList))
{
if (pdList.Count > 1)
{
try
{
//!!! FAILURE !!!
//this fails with out of range exception
var tl = new Task[pdList.Count];
for(int i = 0; i < pdList.Count; i++)
{
//!!! EXCEPTION !!!
// this causes an out of range exception
// at exec task is attempting to read final value of i
// is i accessed by ref? -- no it's a shared resource
tl[i] = Task.Factory.StartNew(() => ExecuteProcess(pdList[i], hargs), TaskCreationOptions.LongRunning);
}
//this does not cause an exception
//var tl = new List<Task>();
//foreach (var pd in pdList)
//{
// tl.Add(Task.Factory.StartNew(() => ExecuteProcess(pd, hargs), TaskCreationOptions.LongRunning));
//}
Task.WaitAll(tl);
// this does not cause an exception
//Parallel.ForEach(pdList, pd => ExecuteProcess(pd, hargs));
}
catch (Exception e)
{
// log exception
}
}
else
{// avoid parallelization if only one process is expected
// to be executed
// execute on calling thread
ExecuteProcess(pdList[0], hargs);
}
}
}
正如 Jon Skeet 指出的那样，这是一个简单的疏忽和睡眠不足。当任务执行时，我是共享资源。
var tl = new Task[pdList.Count];
for (int i = 0; i < pdList.Count; i++)
{
int li = i; // i is shared...use local copy
// use foreach instead
tl[i] = Task.Factory.StartNew(() => ExecuteProcess(pdList[li], hargs), TaskCreationOptions.LongRunning);
}
工作解决方案：
int pi = 0;
var plist = new Task[pdList.Count];
foreach(var pd in pdList)
{
plist[pi++] = Task.Factory.StartNew(() => ExecuteProcess(pd, hargs), TaskCreationOptions.LongRunning);
}
Task.WaitAll(plist);