受CPU限制-上下文切换-慢线程处理

[etl2016]

你好

我的环境如下：具有16个虚拟处理器和64 GB内存的虚拟机。我有一个小的测试csv文件，有十六万行。我正在以每个CPU一个线程的速度旋转16个线程，因此，每个线程分担一万行的工作量。每个线程都将一万行转换为等效的内存数据表。并且，直到此阶段，所有处理都在几秒钟内发生（从程序启动后不到7-10秒，并且在这个时间点上，我有16个内存数据表的集合，每个数据表包含一万行）。因此，磁盘IO在此阶段到此结束。从现在开始，进行CPU绑定处理。 16个线程中的每一个继续循环遍历它们各自的工作量，即万行内存数据表，并连接到数据库以更新其各自内存数据表中的列之一。观察到这万行的总吞吐量为90分钟，大约每分钟100行。 16个线程中的每个线程花费的时间完全相同，并且在这普通的90分钟内，所有16个线程都处理各自的一万行的工作负载。这看起来非常非常低的吞吐量。仔细研究数据库的周转时间，可以证明它很快，数据库可以在几毫秒/微秒内做出响应。循环中没有磁盘IO。在数据库中查找该值后，内存中的数据表将立即更新。而且，每条指令本身的执行速度都非常快（从记录的时间开始）。然而，已经观察到在循环内的连续编程指令之间，存在短的延迟，这可能是由于上下文/线程切换引起的。

问题11：在由16个虚拟处理器和64 GB内存组成的硬件设置中，程序以1：1的比例旋转线程（并且VM中几乎运行了零个用户程序），为什么会发生上下文切换？问题2：异步TPL异步编程可以在这里提供帮助吗-我认为这几乎没有帮助（请更正我），因为在等待数据库响应时程序无需执行任何操作。 for循环中几乎没有3或4条指令来处理一万行内存数据表，并且这些编程指令中的每条指令都取决于前一条编程指令的完成情况，X指令无法执行的所有操作-sync，同时仍在处理其前任指令（X-1）。 for循环中的这3或4条指令必须顺序执行。 X，X + 1和X + 2的发生速度以毫秒/微秒为单位。似乎根本原因是某些X和X + 1或（x-1）和x有时甚至是（X + 2）和（X + 3）之间的短暂延迟，从而使总体吞吐量每秒仅超过1行。

您能否分享您的看法？

谢谢

 

[跳伞]

您必须为每一行访问数据库这一事实仍然表明您受IO约束。更糟糕的是，如果数据库位于网络上，则该IO会受到网络延迟的影响。因此，实际的数据库操作可能非常快，但是将查询/命令发送到数据库并使其返回所需的时间是您可能花费更多的时间运行更多的CPU指令，而不是产生线程并等待线程的时间。响应。

 

[跳伞]

As an aside, is there a chance that you have accidentally left some Console.WriteLine()'s in your code?

 

[跳伞]

我会问一个明显的问题：您的探查器告诉您什么时间在哪里？

 

[etl2016]

".....更糟糕的是，如果数据库位于网络上，则该IO会受到网络延迟的影响....."同意该数据库基于云，但是与我的处理组件位于同一地理区域。因此，客户端服务器的定位不是很明显，因为它本来就属于内部部署，除了知道它们属于同一区域之外。

".....顺便说一句，您是否有可能不小心在代码中留下了一些Console.WriteLine（）？ ...."确实如此。在每个Console.Writeline中，都从Managed-ThreadID和系统时间中捕获线程号。这可能会增加成本。这是为了监视响应时间。我将删除它们，以了解实际的非记录时间有多长时间。将探索替代选项以监控周转时间。

".....您的探查器会告诉您有关时间在哪里？ ...."我将使用此方法作为Console.Writeline的替代选项，尚未使用此功能，非常感谢任何指针。

 

[赫尔曼]

我不认为硬件或网络延迟不是问题。例如，向Azure数据库中插入160K行应该不超过几分钟。

如果您的目标是将CSV导入数据库，则可以使用实体框架连接到数据库。将行导入到本地对象集中，并让EF管理批处理插入。另一种解决方案是使用VALUES动态构建一个（或多个）巨型批处理SQL查询以创建那些160K行。SQL语句受到限制默认情况下最大为10MB，因此您可能必须将最终查询分成大约10K行。这样一来，每行只剩下1KB的查询文本，而实际上只有16条查询可以运行。然后，您可以并行处理并从服务器获得完整的吞吐量。

您绝对不应该尝试做的一件事是一次遍历一个查询...

特别是，如果您在其中进行任何字符串连接，那么它实际上会使整个过程陷入困境...

如果您要发布代码，它将对我们有帮助。

 

[etl2016]

谢谢你赫尔曼。

我在VS2017下使用Performance Profiler检查了运行时行为。".........例如，向Azure数据库中插入16万行应该不超过几分钟。"并非所有这16万行（也不是整行）都插入到目标DB中。我的数据库不是传统的Azure SQL，而是NoSQL Azure Redis缓存，它是一对{key，value}对，具有标准SQL的有限功能。具有一万行的内存数据表逐行循环，并在Redis中查找特定的列，并且a）如果键列存在则检索值-如果尚不存在-b.1）新值对应于要查找的键来构造，并且b.2）在Redis中填充为新的{key，value}对，并且b.3）用相应的新构造的值更新内存中的数据表。

从性能分析中可以看出，a）不到1毫秒。 b.1）花费的时间少于1毫秒b.2）花费的时间最长，大约为400毫秒b.3）花费的时间为20毫秒。最终结果是每秒超过1行且低于2行吞吐量，或每分钟大约100行。这意味着，具有一万行的数据表需要大约一个半小时才能完成Redis上的“获取/设置”周期，并随后使用从{key，value}对中获取的相应值来更新该数据表。

我的.net代码和Azure Redis缓存位于同一地理区域，因此，网络（可能是？）可能不是根本原因。

谢谢

 

[赫尔曼]

就像我说的那样，这将有助于查看代码...我不特别了解Redis，但是从我的阅读中可以看出，这是一个简单的内存缓存。看来支持批处理，看看 这里 对于这样做的图书馆，也许这会有所帮助。

 

[etl2016]

谢谢，我的代码如下：

foreach（万行DataTable.Rows中的DataRow行）//每行都是2列的临时结构，以{key，value = null}开头，然后循环
{
1）var value =从Redis检索密钥的值；
2）if（检索的值为null）//键在Redis中尚不存在
2.1）value =构造一个与正在处理的密钥相对应的新值
2.2）设置新构造的值以在Redis数据库中建立新的{key，value}对
2.3）在万行数据表中更新与键对应的相同的新构造值
3）else //，这意味着Redis已经拥有此{key，value}对
使用从Redis获取的值更新与万行数据表中的键列对应的值
}

在这里，将新构造的值设置为Redis以填充新的{key，value}对的2.2）是最长的运行指令，耗时近400毫秒

如原型 .NET开发人员的Redis – Redis管道批处理|塔斯瓦尔·巴蒂（Taswar Bhatti） ，我正在尝试构造一个批处理，从技术上讲，这是一万个StringGetAsync（key）指令的.net列表，并且在构造完该批处理后，我要执行该批处理。

在这里，我试图了解如何编写.net TPL异步代码，以将请求A与响应A关联。如果Redis持有3对{key1，value1} {key2，value2}和{key3，value3}，那么当一批的"告诉我key1，key2，key3的值"运行，Redis回复"在这里，您可以输入3个值：value3，value1，value2"，问题是如何将value1与key1对应起来？有什么共同点将请求和响应联系在一起，尤其是在一万个请求的foreach循环中？非常感谢任何示例代码（.net FW 4.6），谢谢。

 

[赫尔曼]

不，那不是代码。复制并粘贴实际代码，这样我们就不会无缘无故地绕圈数小时。

 

[羊皮]

坦白地说，在当今时代，人们需要指导他人如何发布代码和使用代码标签。这是您使用论坛编辑器提供的代码标签功能发布代码的方式：

您随附的内容；不是代码。这是最坏形式的伪代码。这很简单。您发布您的实际代码，或者没有理由继续此主题。

我们不需要您的代码在做什么的解释，因为这将由我们决定，这也是您在这里的原因。但是，告诉我们您希望代码执行什么操作可能会更有帮助。

您能否停止这样的报价："别人说了些什么"并在绝对必要时使用引号按钮，然后仅引述与您的回复相关的特定行。否则，将很难阅读什么是引号和什么是文本，我希望不要阅读两次。

谢谢

 

[etl2016]

谢谢Herman，Sheepings。代码如下：

IBatch batch = redis.CreateBatch ();

var 清单ofGetTasks = new 清单 <Task<RedisValue>> ();



foreach (DataRow row in in_memory_Datatable .Rows)

{

   var readTask = batch.StringGetAsync (row["key"] );

   清单ofGetTasks.Add (readTask );

}

batch.Execute();

在这个地方，我试图从Redis捕获与请求相对应的响应，并标识每个请求与其对应响应之间的关联。并且，为每个对应的键更新内存数据表的value列。

例如，如果内存数据表在接收来自Redis的响应之前以{key1，null}，{key2，null}和{key3，null}保留3行，那么最终的预期行为是让in -memory-Datatable具有内容{key1，value1}，{key2，value2}和{key3，value3}

谢谢

 

[赫尔曼]

好的，因此从该代码中，我们看到您正在为内存表的每一行调用StringGetAsync（我认为这是一个远程方法）。

由于它仍然缺少一堆代码，因此我认为您应该能够这样做。使用Redis一目了然地获得整个字典 这个 。将此作为您的工作集。做您的foreach，但不必每次都调用服务，而只需查找本地工作集。

 

[跳伞]

The pseudo code you posted in post #9 does not look like the snippet of code in post #12. Where is the DataTable updates you mentioned in post #1 和 #9? Where is the Redis database updates? Where are the Console.WriteLine()'s you mentioned in post #5?

 

[etl2016]

谢谢跳伞。

现在已删除Console.Writeln，并且正在使用Profiler进行性能监视。此外，该方法正从较早的串行升级为异步，以解决网络往返问题。

串行完成时更新到redis（对不起，最好是添加新的{key，value}对），这是由于响应请求从.net代码到中间StackExchange.Redis库到最终Redis数据库的往返而已确定的瓶颈。 。在此设计中，redis永远不会更新，只有Get / Set。 Get操作的发生速度比Set操作要快得多。

现在，串行获取/设置被重新编码为TPL /异步方法，在该方法中，首先构建并触发了所有万个Get指令中的第1批，并在前面的第7至17行中将其触发。下一步，我试图弄清楚如何将Redis反馈的一万个响应与其各自的请求相关联，以使{key，value}对正确对齐。正在构建实现​​此关联的TPL代码。欢迎任何输入。

一旦正确地关联了请求和响应，内存中的数据表更新就不会以TPL /异步方式发生，但是，如使用较早的逐行同步方法所检查的，此内存中的数据表操作的时间为1行花费的时间比同步设置操作（400毫秒）少得多（10毫秒）。因此，现在的焦点集中在以TPL异步方式将Set方法重写为Pipeline / Batch方法。

作为TPL / Asynch管道批处理编程方式的另一种方法，正在考虑查看Redis是否提供了将所有{key，value}对下载到用户编程运行时环境的功能，因此完全避免了网络带宽问题。

比你

 

[跳伞]

如先前文章中的第7至17行所述，进行了建造和射击。下一步，我试图弄清楚如何将Redis反馈的一万个响应与其各自的请求相关联，以使{key，value}对正确对齐。正在构建实现​​此关联的TPL代码。欢迎任何输入。

我的阅读 文件资料 建议响应以与批处理中请求相同的顺序返回。

至于DataTable的异步更新，我相信DataTable方法是在异步/等待出现之前编写的，并且没有像框架中其他较旧的类那样使用更新的异步任务方法进行过改造。我的直觉是，您最大的收获可能是对批处理结果进行分区并并行更新表行，但这只能使用分析器进行验证。通过触发更多线程，您可能会像在运行数百个线程之前一样，强制进行更多的上下文切换并实际上放慢速度。

 

[赫尔曼]

因此，看了一点之后，似乎Entity Framework 6.0支持Redis上下文。这可能会解决您的性能问题，并提供一个强大的界面来工作Redis。

在这里看看： 使用实体框架6访问Redis数据

和这里： 将Azure Redis缓存与MVC和实体框架一起使用

 

[跳伞]

除非我们的OP对问题的描述不完整，否则根据第1条帖子，他所谈论的数据库是Redis数据库。由于他只是更新一个键值，因此我怀疑围绕Redis使用EF包装器是否真的会对OP起到更大的作用。

I also have some doubts about even needing those DataTable's that he is putting the parsed CSV's into. Nothing in his description indicates that he is cross-referencing the current row with other rows in the DataTable when determining a key, or computing a value if the key is not found in the database. Neither does his pseudo code in post #9, or post #12 show any cross row accesses. If there is no need to reference other rows, then a simple list of rows parsed from the CSV is needed (to give O(1) performance when retrieving an item from the list) as opposed to trying to get a row from the DataTable (which has an O(log n) performance).

 

[etl2016]

嗨，我注意到以下管道方法带来了一些显着的性能提升。

清单 <Task> addTasks = new 清单<Task>();
for (int i = 0; i < In_memory_DataTable.Rows.Count ; i++ )
{
    DataRow row = In_memory_DataTable.Rows[i];
    Task<bool> addAsync = MultiplexerConn.StringSetAsync() );  // StackExchange.Redis library
    addTasks.Add(addAsync);
}

Task[] tasks = addTasks.ToArray();
Task.WaitAll(task);

但是，以上只是评估流水线方法性能的一种尝试，我已经证明它是超快的（不到20分钟即可达到200k），因此我将朝着这个方向继续。

现在，通过上述行之有效的性能证明流水线方法，我正在尝试实现我的实际功能需求，这是基于"concurrency"建议在本文结尾处关于StackExchange.Redis库： 管道和多路复用器

正在研究的方法是：

1）使用StackExchange.Redis库方法StringGetAsync异步获取与我的Key相对应的Value（其中key是我的内存Datatable中的每个元素，大约1万个单元）（以及16个此类线程同时处理它们各自的10k Datatable工作负载）
2）如果检索成功，请使用获取的值，更新我的内存DataTable
3）如果不是，请构造一个新值，并使用StackExchange.Redis库方法StringSetAsync（键，值）在redis中设置新的{key，value}对。

从研究中可以理解，等待是比继续执行更有效的方法，并且正在考虑使用此处建议的方法： 完成异步任务后处理.

由于我的C＃程序和端点Redis数据库之间的通信必须通过StackExchange.Redis库进行，因此，我正在探索StackExchange的StringGetAsync和StringSetAsync方法在后台已实现了哪些TPL / Async功能（例如启动Task。运行），这样就不必（不应）在我的程序代码中冗余地实现这些功能。

欢迎您提供任何其他方法的建议或建议，谢谢。

 

[赫尔曼]

硬着头皮，安装了磁带库，并设法在5分钟内提出了建议。一次性完成所有操作，您可以使用When和CommandFlags枚举来管理行为...如果需要响应，则可以“等待”响应并删除FireAndForget标志...如果需要检索之前或之后的值，之后，使用相同的重载，并使用StringGet返回一个数组。不要遍历它，让库立即获取并设置它们。与往常一样，最好让图书馆做自己的事情。

附带说明，安装库后，我意识到您再次发布了不完整的代码...我不明白什么很难理解，如果您需要解决某些代码问题的帮助，则应将确切的代码发布在第一名。不是复制品，示例，原型或工作流程。确切的代码。否则，您会要求所有人阅读您的帖子，并试图帮助您加倍努力以获取他们需要的信息来回答您的问题。没考虑

来吧，人们在一起。

        public void SetRedisValues(DataTable table, string colKey, string colValue)
        {
            using (var mux = ConnectionMultiplexer.Connect("localhost"))
            {
                var db = mux.GetDatabase();

                KeyValuePair<RedisKey, RedisValue>[] values = table.Rows.Cast<DataRow>()
                                                                   .ToDictionary(r => new RedisKey(r[colKey].ToString()),
                                                                                 r => new RedisValue(r[colValue].ToString()),
                                                                                 null)
                                                                   .ToArray();

                var task = db.StringSetAsync(values, When.Always, CommandFlags.FireAndForget);
            }
        }