44

我正在尝试从子进程返回值，但不幸的是这些值是不可取的。因此，我在线程模块中成功使用了全局变量，但在使用多处理模块时无法检索子进程中完成的更新。我希望我错过了什么。

最后打印的结果总是与给定 varsdataDV03和dataDV04. 子进程正在更新这些全局变量，但这些全局变量在父进程中保持不变。

import multiprocessing

# NOT ABLE to get python to return values in passed variables.

ants = ['DV03', 'DV04']
dataDV03 = ['', '']
dataDV04 = {'driver': '', 'status': ''}


def getDV03CclDrivers(lib):  # call global variable
    global dataDV03
    dataDV03[1] = 1
    dataDV03[0] = 0

# eval( 'CCL.' + lib + '.' +  lib + '( "DV03" )' ) these are unpicklable instantiations

def getDV04CclDrivers(lib, dataDV04):   # pass global variable
    dataDV04['driver'] = 0  # eval( 'CCL.' + lib + '.' +  lib + '( "DV04" )' )


if __name__ == "__main__":

    jobs = []
    if 'DV03' in ants:
        j = multiprocessing.Process(target=getDV03CclDrivers, args=('LORR',))
        jobs.append(j)

    if 'DV04' in ants:
        j = multiprocessing.Process(target=getDV04CclDrivers, args=('LORR', dataDV04))
        jobs.append(j)

    for j in jobs:
        j.start()

    for j in jobs:
        j.join()

    print 'Results:\n'
    print 'DV03', dataDV03
    print 'DV04', dataDV04

我无法发布我的问题，因此将尝试编辑原件。

这是不可腌制的对象：

In [1]: from CCL import LORR
In [2]: lorr=LORR.LORR('DV20', None)
In [3]: lorr
Out[3]: <CCL.LORR.LORR instance at 0x94b188c>

这是我使用 multiprocessing.Pool 将实例返回给父级时返回的错误：

Thread getCcl (('DV20', 'LORR'),)
Process PoolWorker-1:
Traceback (most recent call last):
File "/alma/ACS-10.1/casa/lib/python2.6/multiprocessing/process.py", line 232, in _bootstrap
self.run()
File "/alma/ACS-10.1/casa/lib/python2.6/multiprocessing/process.py", line 88, in run
self._target(*self._args, **self._kwargs)
File "/alma/ACS-10.1/casa/lib/python2.6/multiprocessing/pool.py", line 71, in worker
put((job, i, result))
File "/alma/ACS-10.1/casa/lib/python2.6/multiprocessing/queues.py", line 366, in put
return send(obj)
UnpickleableError: Cannot pickle <type 'thread.lock'> objects

In [5]: dir(lorr)
Out[5]:
['GET_AMBIENT_TEMPERATURE',
 'GET_CAN_ERROR',
 'GET_CAN_ERROR_COUNT',
 'GET_CHANNEL_NUMBER',
 'GET_COUNT_PER_C_OP',
 'GET_COUNT_REMAINING_OP',
 'GET_DCM_LOCKED',
 'GET_EFC_125_MHZ',
 'GET_EFC_COMB_LINE_PLL',
 'GET_ERROR_CODE_LAST_CAN_ERROR',
 'GET_INTERNAL_SLAVE_ERROR_CODE',
 'GET_MAGNITUDE_CELSIUS_OP',
 'GET_MAJOR_REV_LEVEL',
 'GET_MINOR_REV_LEVEL',
 'GET_MODULE_CODES_CDAY',
 'GET_MODULE_CODES_CMONTH',
 'GET_MODULE_CODES_DIG1',
 'GET_MODULE_CODES_DIG2',
 'GET_MODULE_CODES_DIG4',
 'GET_MODULE_CODES_DIG6',
 'GET_MODULE_CODES_SERIAL',
 'GET_MODULE_CODES_VERSION_MAJOR',
 'GET_MODULE_CODES_VERSION_MINOR',
 'GET_MODULE_CODES_YEAR',
 'GET_NODE_ADDRESS',
 'GET_OPTICAL_POWER_OFF',
 'GET_OUTPUT_125MHZ_LOCKED',
 'GET_OUTPUT_2GHZ_LOCKED',
 'GET_PATCH_LEVEL',
 'GET_POWER_SUPPLY_12V_NOT_OK',
 'GET_POWER_SUPPLY_15V_NOT_OK',
 'GET_PROTOCOL_MAJOR_REV_LEVEL',
 'GET_PROTOCOL_MINOR_REV_LEVEL',
 'GET_PROTOCOL_PATCH_LEVEL',
 'GET_PROTOCOL_REV_LEVEL',
 'GET_PWR_125_MHZ',
 'GET_PWR_25_MHZ',
 'GET_PWR_2_GHZ',
 'GET_READ_MODULE_CODES',
 'GET_RX_OPT_PWR',
 'GET_SERIAL_NUMBER',
 'GET_SIGN_OP',
 'GET_STATUS',
 'GET_SW_REV_LEVEL',
 'GET_TE_LENGTH',
 'GET_TE_LONG_FLAG_SET',
 'GET_TE_OFFSET_COUNTER',
 'GET_TE_SHORT_FLAG_SET',
 'GET_TRANS_NUM',
 'GET_VDC_12',
 'GET_VDC_15',
 'GET_VDC_7',
 'GET_VDC_MINUS_7',
 'SET_CLEAR_FLAGS',
 'SET_FPGA_LOGIC_RESET',
 'SET_RESET_AMBSI',
 'SET_RESET_DEVICE',
 'SET_RESYNC_TE',
 'STATUS',
 '_HardwareDevice__componentName',
 '_HardwareDevice__hw',
 '_HardwareDevice__stickyFlag',
 '_LORRBase__logger',
 '__del__',
 '__doc__',
 '__init__',
 '__module__',
 '_devices',
 'clearDeviceCommunicationErrorAlarm',
 'getControlList',
 'getDeviceCommunicationErrorCounter',
 'getErrorMessage',
 'getHwState',
 'getInternalSlaveCanErrorMsg',
 'getLastCanErrorMsg',
 'getMonitorList',
 'hwConfigure',
 'hwDiagnostic',
 'hwInitialize',
 'hwOperational',
 'hwSimulation',
 'hwStart',
 'hwStop',
 'inErrorState',
 'isMonitoring',
 'isSimulated']

In [6]:
4

5 回答 5

51

当您使用multiprocessing打开第二个进程时，将创建一个具有自己全局状态的全新 Python 实例。该全局状态不是共享的，因此子进程对全局变量所做的更改对父进程是不可见的。

此外，大多数multiprocessing提供使用 pickle 的抽象来传输数据。使用代理传输的所有数据都必须是可腌制的；其中包括 aManager提供的所有对象。相关引文（我的重点）：

确保代理方法的参数是可挑选的。

并且（在该Manager部分中）：

其他进程可以使用代理访问共享对象。

Queues 还需要可腌制的数据；文档没有这么说，但快速测试证实了这一点：

import multiprocessing
import pickle

class Thing(object):
    def __getstate__(self):
        print 'got pickled'
        return self.__dict__
    def __setstate__(self, state):
        print 'got unpickled'
        self.__dict__.update(state)

q = multiprocessing.Queue()
p = multiprocessing.Process(target=q.put, args=(Thing(),))
p.start()
print q.get()
p.join()

输出：

$ python mp.py 
got pickled
got unpickled
<__main__.Thing object at 0x10056b350>

如果您真的无法腌制数据，那么可能对您有用的一种方法是找到一种方法将其存储为ctype对象；然后可以将对内存的引用传递给子进程。这对我来说似乎很狡猾。我从来没有做过。但这对您来说可能是一个可能的解决方案。

鉴于您的更新，您似乎需要更多地了解LORR. 是LORR班级吗？你能继承它吗？它是其他东西的子类吗？它的 MRO 是什么？（如果可行，请尝试LORR.__mro__发布输出。）如果它是纯 python 对象，则可以对其进行子类化，创建 a__setstate__和 a__getstate__以启用酸洗。

另一种方法可能是弄清楚如何从LORR实例中获取相关数据并通过简单的字符串传递它。既然你说你真的只是想调用对象的方法，那为什么不直接用Queues 来回发送消息呢？换句话说，像这样（示意性地）：

Main Process              Child 1                       Child 2
                          LORR 1                        LORR 2 
child1_in_queue     ->    get message 'foo'
                          call 'foo' method
child1_out_queue    <-    return foo data string
child2_in_queue                   ->                    get message 'bar'
                                                        call 'bar' method
child2_out_queue                  <-                    return bar data string
于 2012-06-15T18:39:33.197 回答
7

@DBlas 在答案中为您提供了一个快速 url 和对 Manager 类的引用，但我认为它仍然有点模糊，所以我认为它可能对您有所帮助。

import multiprocessing
from multiprocessing import Manager

ants = ['DV03', 'DV04']

def getDV03CclDrivers(lib, data_dict):  
    data_dict[1] = 1
    data_dict[0] = 0

def getDV04CclDrivers(lib, data_list):   
    data_list['driver'] = 0  


if __name__ == "__main__":

    manager = Manager()
    dataDV03 = manager.list(['', ''])
    dataDV04 = manager.dict({'driver': '', 'status': ''})

    jobs = []
    if 'DV03' in ants:
        j = multiprocessing.Process(
                target=getDV03CclDrivers, 
                args=('LORR', dataDV03))
        jobs.append(j)

    if 'DV04' in ants:
        j = multiprocessing.Process(
                target=getDV04CclDrivers, 
                args=('LORR', dataDV04))
        jobs.append(j)

    for j in jobs:
        j.start()

    for j in jobs:
        j.join()

    print 'Results:\n'
    print 'DV03', dataDV03
    print 'DV04', dataDV04

因为多处理实际上使用单独的进程，所以不能简单地共享全局变量，因为它们将位于内存中完全不同的“空间”中。您在一个流程下对全局所做的事情不会反映在另一个流程中。虽然我承认它看起来很混乱，因为你看到它的方式，它都存在于同一段代码中，所以“为什么这些方法不能访问全局”？很难理解它们将在不同进程中运行的想法。

Manager 类用作数据结构的代理，可以在进程之间为您来回传递信息。您将要做的是从经理创建一个特殊的字典和列表，将它们传递到您的方法中，并在本地对其进行操作。

不可腌制的数据

对于您的专业 LORR 对象，您可能需要创建类似代理的东西，它可以表示实例的可选择状态。

不是超级健壮或经过太多测试，但会给你这个想法。

class LORRProxy(object):

    def __init__(self, lorrObject=None):
        self.instance = lorrObject

    def __getstate__(self):
        # how to get the state data out of a lorr instance
        inst = self.instance
        state = dict(
            foo = inst.a,
            bar = inst.b,
        )
        return state

    def __setstate__(self, state):
        # rebuilt a lorr instance from state
        lorr = LORR.LORR()
        lorr.a = state['foo']
        lorr.b = state['bar']
        self.instance = lorr
于 2012-06-15T18:33:46.380 回答
6

使用时multiprocess，进程之间传递对象的唯一方法是使用Queueor Pipe；全局变量不共享。对象必须是可腌制的，所以multiprocess在这里不会帮助你。

于 2012-06-15T18:24:15.237 回答
5

您还可以使用多处理 Array。这允许您在进程之间拥有共享状态，并且可能是最接近全局变量的东西。

在 main 的顶部，声明一个数组。第一个参数“i”表示它将是整数。第二个参数给出初始值：

shared_dataDV03 = multiprocessing.Array ('i', (0, 0)) #a shared array

然后将此数组作为参数传递给进程：

j = multiprocessing.Process(target=getDV03CclDrivers, args=('LORR',shared_dataDV03))

您必须在被调用的函数中接收数组参数，然后您可以在函数中对其进行修改：

def getDV03CclDrivers(lib,arr):  # call global variable
    arr[1]=1
    arr[0]=0

该数组与父级共享，因此您可以在父级中打印出末尾的值：

print 'DV03', shared_dataDV03[:]

它将显示更改：

DV03 [0, 1]
于 2016-09-08T07:14:22.843 回答
1

我使用 p.map() 将多个进程分拆到远程服务器，并在它们在不可预测的时间返回时打印结果：

Servers=[...]
from multiprocessing import Pool
p=Pool(len(Servers))
p.map(DoIndividualSummary, Servers)

如果DoIndividualSummary用于print结果，这工作得很好，但整体结果的顺序不可预测，这使得解释变得困难。我尝试了多种使用全局变量的方法，但都遇到了问题。最后，我用 sqlite3 成功了。

之前p.map()，打开一个 sqlite 连接并创建一个表：

import sqlite3
conn=sqlite3.connect('servers.db') # need conn for commit and close
db=conn.cursor()
try: db.execute('''drop table servers''')
except: pass
db.execute('''CREATE TABLE servers (server text, serverdetail text, readings     text)''')
conn.commit()

然后，当从 中返回时DoIndividualSummary()，将结果保存到表中：

db.execute('''INSERT INTO servers VALUES (?,?,?)''',         (server,serverdetail,readings))
conn.commit()
return

语句后map()，打印结果：

db.execute('''select * from servers order by server''')
rows=db.fetchall()
for server,serverdetail,readings in rows: print serverdetail,readings

可能看起来有点矫枉过正，但对我来说它比推荐的解决方案更简单。

于 2014-10-14T11:14:23.087 回答