我正在训练一个典型的map-reduce架构(在OS类中),我可以自由决定主进程如何告诉其N子进程解析日志.所以,我有点陷入这两种可能性:
计算行数并为每个映射OR提供X行
每个地图读取其ID的行,下一行读取= current_one + number_of_existent_maps例如:有3个地图,每个地图将读取这些行:
我必须执行此操作才能超出分析整个日志文件的单个进程,因此我在子进程之间拆分作业的方式必须与此目标一致.
你觉得哪一个最好？如何使scanf或fgets适应1)或2)？
我会对2)的一些示例代码感到满意,因为fork/pipes不是我的问题:P
重新编辑: 我不鼓励在这里使用select,只在map procs和将监视读取的reduce进程之间.我现在有限制:
我希望每个进程都读取total_lines/N行.但似乎我必须让map procs打开文件然后读取相应的行.所以这是我的疑惑:
1-使每个过程同时或几乎同时打开文件是不是甚至可能？这有助于加快速度吗？
2-如果不可能这样做,我将让一个父文件打开文件(而不是每个孩子这样做)发送一个具有最小和最大限制的结构,然后map procs将读取他们负责的任何行for,处理它们并给予reduce过程(这对于现在的问题无关紧要).
如何通过N个地图正确划分行数并同时将它们读取？我认为fseek()可能是一个好武器,但我不知道我如何使用它.请帮忙!
如果我理解正确,您希望让所有进程从单个文件中读取行.我不推荐这个,它有点混乱,你必须a)多次读取文件的相同部分或b)使用锁定/互斥或其他一些机制来避免这种情况.它会变得复杂而且难以调试.
我有一个主进程读取该文件,并将行分配给子进程池.您可以使用共享内存来加快速度,并减少对数据复制IPC的需求; 或使用线程.
至于示例,我回答了一个关于分叉和IPC的问题,并给出了一个示例函数的代码片段,该函数分叉并返回一对管道以进行父子通信.让我看一下(...)这里是= P Can popen()制作像pipe()+ fork()这样的双向管道？
编辑:我一直在想这个= P. 这是一个想法:
有一个主进程产生子进程,其类似于我在上面链接中显示的内容.
每个进程首先向主机发送一个字节,以表示它已准备就绪,然后阻塞read().
让主进程从文件读取一行到共享内存缓冲区,并阻塞select()其子管道.
当select()返回时,读取信号和准备发送到子在共享存储器空间中的线的偏移的字节之一.
主进程重复(读取一行,选择块,读取一个字节以消耗就绪事件等)
子节点以您需要的任何方式处理线路,然后向主节点发送一个字节以再次发出准备信号.
(如果需要,你可以避免使用共享内存缓冲区,并将线路发送到管道中,尽管它会涉及不断的数据复制.如果每条线路的处理计算成本很高,那么它不会产生任何影响;但是如果线条需要很少的处理,它可能会减慢你的速度).
我希望这有帮助!
根据Newba的评论编辑2:
好的,所以没有共享内存.使用上面的模型,只代替向下发送管道读取共享内存空间中读取的行的偏移量,发送整行.这可能听起来像你在浪费时间,只能从文件中读取它,但相信我,你不是.管道比从硬盘中的常规文件读取快几个数量级,如果您希望子进程直接从文件读取,您将遇到我在答案开头指出的问题.
那么,掌握过程:
Spawn子进程使用类似我写的函数(上面的链接),为双向通信创建管道.
从文件中读取一行到缓冲区(私有,本地,无共享内存).
您现在可以处理数据了.调用select()来阻止与子进程通信的所有管道.
选择任何有可用数据的管道,从中读取一个字节,然后将等待处理的行发送到相应管道的缓冲区中(记住,我们每个子进程有2个,上升,一个往下走).
从步骤2开始重复,即读取另一行.
子进程:
当他们开始时,他们有一个阅读管和一个书写管可供他们使用.向写入管道发送一个字节,以指示您准备好的主进程并等待数据处理(这是我们在上面的步骤4中读取的单个字节).
阻止read(),等待主进程(知道您已准备就绪,因为步骤1)将数据发送到进程.继续阅读,直到你到达换行符(你说你正在读行,对吧？).注意我正在关注您的模型,一次向每个进程发送一行,如果您愿意,可以发送多行.
处理数据.
返回步骤1,即发送另一个字节,表示您已准备好接收更多数据.
你可以使用简单的协议将任务分配给任意数量的子进程.对一个孩子,n个孩子(其中n是计算机中的核心数)和超过n个孩子进行测试可能会很有趣,并比较性能.
哇,这是一个很长的答案.我真的希望我能帮助xD
| 归档时间： |
|
| 查看次数： |
686 次 |
| 最近记录： |