创建表示3位二进制字符串的所有组合的最小图

Question

我有一个算法,创建一个图形,其中包含以最短图形路径的形式编码的3位二进制字符串的所有表示,其中路径中的偶数表示0,而奇数表示1:

from itertools import permutations, product
import networkx as nx
import progressbar
import itertools

def groups(sources, template):
    func = permutations
    keys = sources.keys()
    combos = [func(sources[k], template.count(k)) for k in keys]
    for t in product(*combos):
        d = {k: iter(n) for k, n in zip(keys, t)}
        yield [next(d[k]) for k in template]                                      

g = nx.Graph()

added = []   
good = []
index = []
# I create list with 3-bit binary strings
# I do not include one of the pairs of binary strings that have a mirror image
list_1 = [list(i) for i in itertools.product(tuple(range(2)), repeat=3) if tuple(reversed(i)) >= tuple(i)]
count = list(range(len(list_1)))

h = 0
while len(added) < len(list_1): 
     # In each next step I enlarge the list 'good` by the next even and odd number
     if h != 0:   
        for q in range(2):   
            good.append([i for i in good if i%2 == q][-1] + 2)
     # I create a list `c` with string indices from the list` list_1`, that are not yet used.
     # Whereas the `index` list stores the numbering of strings from the list` list_1`, whose representations have already been correctly added to the `added` list.          
     c = [item for item in count if item not in index]

     for m in c:
     # I create representations of binary strings, where 0 is 'v0' and 1 is 'v1'. For example, the '001' combination is now 'v0v0v1'
         a = ['v{}'.format(x%2) for x in list_1[m]]

         if h == 0:
            for w in range(2):
                if len([i for i in good if i%2 == w]) < a.count('v{}'.format(w)):
                   for j in range(len([i for i in good if i%2 == w]), a.count('v{}'.format(w))):
                       good.insert(j,2*j + w)                       

         sources={}
         for x in range(2):
             sources["v{0}".format(x)] = [n for n in good if n%2 == x]
         # for each representation in the form 'v0v0v1' for example, I examine all combinations of strings where 'v0' is an even number 'a' v1 'is an odd number, choosing values from the' dobre2 'list and checking the following conditions.
         for aaa_binary in groups(sources, a):             
             # Here, the edges and nodes are added to the graph from the combination of `aaa_binary` and checking whether the combination meets the conditions. If so, it is added to the `added` list. If not, the newly added edges are removed and the next `aaa_binary` combination is taken.           
             g.add_nodes_from (aaa_binary)
             t1 = (aaa_binary[0],aaa_binary[1])
             t2 = (aaa_binary[1],aaa_binary[2]) 

             added_now = []                      
             for edge in (t1,t2):
                 if not g.has_edge(*edge):
                    g.add_edge(*edge)
                    added_now.append(edge)

             added.append(aaa_binary)
             index.append(m)

             for f in range(len(added)):
                 if nx.shortest_path(g, aaa_binary[0], aaa_binary[2]) != aaa_binary or nx.shortest_path(g, added[f][0], added[f][2]) != added[f]:
                    for edge in added_now:
                        g.remove_edge(*edge)
                    added.remove(aaa_binary)
                    index.remove(m)
                    break
             # Calling a good combination search interrupt if it was found and the result added to the `added` list, while the index from the list 'list_1` was added to the` index` list              
             if m in index:
                break

     good.sort()
     set(good)
     index.sort() 

     h = h+1

表示3位二进制字符串的输出路径added:

[[0, 2, 4], [0, 2, 1], [2, 1, 3], [1, 3, 5], [0, 3, 6], [3, 0, 7]]

所以这些是3位二进制字符串的表示:

[[0, 0, 0], [0, 0, 1], [0, 1, 1], [1, 1, 1], [0, 1, 0], [1, 0, 1]]

在步骤h = 0中找到前4个子列表的位置,并在该步骤h = 1中添加最后两个子列表.

当然,正如您所看到的,镜像字符串没有反射,因为在无向图中没有这种需要.

图形:

上述解决方案创建了一个最小的图形并具有唯一的最短路径.这意味着二进制字符串的一个组合在图中仅以最短路径的形式具有一个表示.因此,给定路径的选择是给定二进制序列的单指示.

现在我想在for m in c循环中使用多处理,因为查找元素的顺序在这里无关紧要.

我尝试以这种方式使用多处理:

from multiprocessing import Pool

added = []

def foo(i):
    added = []
       # do something 
       added.append(x[i])
    return added

if __name__ == '__main__':

h = 0
while len(added)<len(c): 

   pool = Pool(4)
   result = pool.imap_unordered(foo, c)      
   added.append(result[-1])

   pool.close()
   pool.join()

   h = h + 1

多处理在while循环中进行,在foo函数中,
added创建列表.在h循环的每个后续步骤中,列表added应按后续值递增,并且added应在函数中使用当前列表foo.是否可以在循环的每个后续步骤中将列表的当前内容传递给函数？因为在上面的代码中,该foo函数added每次都从头开始创建列表的新内容.怎么解决这个问题？

结果导致了不好的结果:

[[0, 2, 4], [0, 2, 1], [2, 1, 3], [1, 3, 5], [0, 1, 2], [1, 0, 3]]

因为对于这样的曲线图中,节点和边,所述条件未得到满足的是nx.shortest_path (graph, i, j) == added[k]对每一个最终节点i, j从added[k] for k in added list.

对于h = 0元素[0, 2, 4], [0, 2, 1], [2, 1, 3], [1, 3, 5]是好的,而在步骤中添加的元素h = 1,即[0, 1, 2], [1, 0, 3]显然在不影响前一步骤的元素的情况下被发现.

怎么解决这个问题？

我意识到这是一种顺序算法,但我也对部分解决方案感兴趣,即甚至算法的部分并行处理.例如,hwhile循环的步骤按顺序运行,但for m in c循环是多处理的.或者其他部分解决方案将改进整个算法以获得更大的组合.

我将非常感谢在我的算法中展示和实现多处理的一些想法.

Answer 1

我认为您不能像现在这样并行化代码.您希望并行化的部分,for m in c循环访问三个全局列表good,added以及index图表g本身.您可以使用a multiprocessing.Array作为列表,但这会破坏整个并行化点,因为multiprocessing.Array(docs)是同步的,因此这些进程实际上并不是并行运行的.

因此,代码需要重构.我首选的并行算法的方法是使用一种生产者/消费者模式

1. 初始化以设置需要执行的作业队列(按顺序运行)
2. 有一个工作池,所有工作人员从该队列中拉出作业(并行运行)
3. 在作业队列耗尽后,聚合结果并构建最终解决方案(按顺序运行)

在这种情况下,1.将是设置代码list_1,count可能是这种h == 0情况.之后你将构建一个" 工作订单 " 队列,这将是c列表 - >将该列表传递给一堆工人 - >获取结果并汇总.问题是for m in c循环的每次执行都可以访问全局状态,并且每次迭代后全局状态都会发生变化.这在逻辑上意味着您无法并行运行代码,因为第一次迭代会更改全局状态并影响第二次迭代的作用.也就是说,根据定义,顺序算法.您不能(至少不容易)并行化迭代构建图形的算法.

你可以使用multiprocessing.starmap和multiprocessing.Array,但这并不能解决问题.您仍然拥有g在所有进程之间共享的图表.因此整个事情需要以这样的方式进行重构:循环上的每次迭代for m in c都独立于该循环的任何其他迭代,或者必须改变整个逻辑,以便for m in c不需要循环开始.

UPDATE

我当时认为你可以通过以下更改将算法转向稍微不那么顺序的版本.我很确定代码已经做了类似的事情,但代码对我来说有点过于密集,而图算法并不是我的专长.

目前,对于新的三元组('101'例如),您将在现有图形中生成所有可能的连接点,然后将新三元组添加到图形中,并根据测量最短路径消除节点.这需要检查图表上的最短路径并进行修改,这会阻止并行化.

注意:以下是关于如何重构代码的相当粗略的大纲,我没有测试过这个或者在数学上验证它实际上是否正常工作

注2:在下面的讨论'101'(注意引号''是一个二进制串,因此是'00'和'1'其中如1,0,4等(没有引号)是图中的顶点的标签.

如果您在现有图表上进行某种子字符串搜索,我将使用第一个三元组作为示例.初始化

1. 生成job_queue包含所有三元组的
2. 占据第一位,并插入,例如,'000'这将是(0,2,4) -这是小事不需要检查任何东西,因为图是空的,当你开始这样的最短路径是定义你插入一个.

在这一点上,你也有部分路径'011','001','010'反之('110'以及'001'因为是无向图).我们将利用现有图表包含剩余三元组的子解决方案这一事实job_queue.让我们说下一个三元组'010',你迭代二进制字符串'010'或list('010')

• 如果图形中'0'已经存在路径/顶点- >继续
• 如果图中'01'已存在路径/顶点- >继续
• 如果存在路径/顶点'010',则不需要添加任何东西(这实际上是一个失败的情况:'010'不应该再进入作业队列,因为它已经解决了).

第二个项目符号点将失败,因为'01'图表中不存在.插入'1'在这种情况下将是1图形的节点并将其连接到三个even节点之一,我不认为哪个是重要的,但你必须记录它连接到哪一个,让我们说你选择了0.该图现在看起来像

 0 - 2 - 4
  \  * 
   \ * 
    \*
     1

最佳边缘以完成路径1 - 2(标有星),以获得一个路径0 - 1 - 2为'010',这是最大化编码三元组的数目,如果边缘的路径1-2添加到图表.如果你添加1-4你只编码'010'三元组,1 - 2编码,'010'但也'001'和'100'.

顺便说一句,我们假设你连接1到2在第一,而不是0(第一连接被挑随机),你现在有一个图

 0 - 2 - 4
     | 
     | 
     |
     1

并且您可以连接1到其中任何一个4或者0,但是您再次获得一个图表,该图表编码剩余的最大三元组数job_queue.

那么如何检查潜在新路径编码的三元组数？您可以相对轻松地检查这一点,更重要的是,可以并行完成检查而无需修改图形g,对于3位字符串,并行节省的成本并不大,但对于32位字符串,它们将是.这是它的工作原理.

1. (顺序)从子路径生成所有可能的完整路径0-1- > (0-1-2), (0-1-4).
2. (并行)对于每个潜在的完整路径,检查路径解决了多少其他三元组,即每个路径候选者生成图表解决的所有三元组并检查这些三元组是否仍然存在job_queue.
   1. (0-1-2)解决了另外两个三元组'001' (4-2-1) or (2-0-1)和'100' (1-0-2) or (1-2-4).
   2. (0-1-4)只解决了三重'010',即本身

解决剩余最多三元组的边缘/路径job_queue是最佳解决方案(我没有证据).

您在2.上面并行运行将图形复制到每个工作人员.因为您没有修改图形,只检查它解决了多少三元组,所以您可以并行执行此操作.每个工人都应该有类似的签名

check_graph(g, path, copy_of_job_queue):
    # do some magic
    return (n_paths_solved, paths_solved)

path是(0-1-2)或者(0-1-4),copy_of_job_queue应该是其余路径的副本job_queue.对于K工作者,您可以创建队列的K个副本.一旦工作池完成,您就知道哪条路径(0-1-2)或(0-1-4)解决了最多的三元组.

您再添加路径和修改图形,并从作业队列中的解决路径.

RINSE - REPEAT直到作业队列为空.

上面有一些明显的问题,对于你进行大量复制和循环的问题job_queue,如果你处理的是大比特空间,比如32比特,那么job_queue很长,所以你可能不想继续复制到所有工人.

对于上面的并行步骤(2.),您可能希望job_queue实际上是dict键是三元组的位置,比如说'010',值是一个布尔标志,表示该三元组是否已经在图形中编码.