使用多处理时截断文件头答案

【问题标题】：Truncated file header while using multiprocessing使用多处理时截断文件头
【发布时间】：2017-06-16 09:48:00
【问题描述】：

当我跑线时：

def book_processing(pair, pool_length):
    p = Pool(len(pool_length)*3)
    temp_parameters = partial(book_call_mprocess, pair)
    p.map_async(temp_parameters, pool_length).get(999999)
    p.close()                                 
    p.join()
    return exchange_books

我收到以下错误：

Traceback (most recent call last):
  File "test_code.py", line 214, in <module>
    current_books = book_call.book_processing(cp, book_list)
  File "/home/user/Desktop/book_call.py", line 155, in book_processing
    p.map_async(temp_parameters, pool_length).get(999999)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 567, in get
    raise self._value
zipfile.BadZipfile: Truncated file header

我觉得好像有一些正在使用的资源在最后一个循环期间没有关闭，但我不确定如何关闭它（仍在学习多处理库）。仅当我的代码相对较快地重复此部分时（在同一分钟内），才会出现此错误。这种情况并不经常发生，但在发生时很明显。

编辑（添加 book_call 代码）：

def book_call_mprocess(currency_pair, ex_list):

    polo_error = 0
    live_error = 0
    kraken_error = 0
    gdax_error = 0

    ex_list = set([ex_list])

    ex_Polo = 'Polo'
    ex_Live = 'Live'
    ex_GDAX = 'GDAX'
    ex_Kraken = 'Kraken'

    cp_polo = 'BTC_ETH'
    cp_kraken = 'XETHXXBT'
    cp_live = 'ETH/BTC'
    cp_GDAX = 'ETH-BTC'

    # Instances
    polo_instance = poloapi.poloniex(polo_key, polo_secret)
    fookraken = krakenapi.API(kraken_key, kraken_secret)
    publicClient = GDAX.PublicClient()

    flag = False
    while not flag:
        flag = False
        err = False

        # Polo Book

        try:
            if ex_Polo in ex_list:
                polo_books = polo_instance.returnOrderBook(cp_polo)
                exchange_books['Polo'] = polo_books
        except:
            err = True
            polo_error = 1

        # Livecoin

        try:
            if ex_Live in ex_list:
                method = "/exchange/order_book"
                live_books = OrderedDict([('currencyPair', cp_live)])
                encoded_data = urllib.urlencode(live_books)
                sign = hmac.new(live_secret, msg=encoded_data, digestmod=hashlib.sha256).hexdigest().upper()
                headers = {"Api-key": live_key, "Sign": sign}
                conn = httplib.HTTPSConnection(server)
                conn.request("GET", method + '?' + encoded_data, '', headers)
                response = conn.getresponse()
                live_books = json.load(response)
                conn.close()
                exchange_books['Live'] = live_books
        except:
            err = True
            live_error = 1

        # Kraken

        try:
            if ex_Kraken in ex_list:
                kraken_books = fookraken.query_public('Depth', {'pair': cp_kraken})
                exchange_books['Kraken'] = kraken_books
        except:
            err = True
            kraken_error = 1

        # GDAX books

        try:
            if ex_GDAX in ex_list:
                gdax_books = publicClient.getProductOrderBook(level=2, product=cp_GDAX)
                exchange_books['GDAX'] = gdax_books
        except:
            err = True
            gdax_error = 1

        flag = True
        if err:
            flag = False
            err = False
            error_list = ['Polo', polo_error, 'Live', live_error, 'Kraken', kraken_error, 'GDAX', gdax_error]
            print_to_excel('excel/error_handler.xlsx', 'Book Call Errors', error_list)
            print "Holding..."
            time.sleep(30)
        return exchange_books


def print_to_excel(workbook, worksheet, data_list):
    ts = str(datetime.datetime.now()).split('.')[0]
    data_list = [ts] + data_list
    wb = load_workbook(workbook)
    if worksheet == 'active':
        ws = wb.active
    else:
        ws = wb[worksheet]
    ws.append(data_list)
    wb.save(workbook)

【问题讨论】：

对我来说，这听起来更像是不同的进程共享一个文件路径（例如临时目录中的一个 zipfile），一个在未完全写入的同时写入另一个读取。你能在book_processing函数中分享处理zipfile的代码吗？
我不确定 zipfile 的作用。我写了一个excel文件，可能就是这样。我已经通过编辑添加了代码。我还添加了我的 print_to_excel 函数。
File "test_code.py", line 214, in <module> 是哪一个？并且：你能纠正缩进吗？（最好粘贴原始代码，然后选择它并按 ctrl-k）
test_code 上的那一行调用 book_processing 函数并传入一个字符串和一个列表。现在更正缩进。
是的，看起来print_to_excel('excel/error_handler.xlsx') 是问题所在，因为它始终在同一位置写入/读取错误文件。如果两个进程同时运行它，你会遇到一个读取 zip 文件而另一个写入它的竞争条件。由于 xlsx 只是 zip 容器中的一堆 xml 文件，这很可能是错误的来源

标签： python python-2.7 error-handling multiprocessing truncated

【解决方案1】：

问题出在函数print_to_excel

更具体地说，在这里：

wb = load_workbook(workbook)

如果两个进程同时运行此函数，您将遇到以下竞态条件：

进程 1 想要打开 error_handler.xlsx，因为它不存在它会创建一个空文件
进程 2 想要打开 error_handler.xlsx，它确实存在，所以它尝试读取它，但它仍然是空的。由于xlsx 格式只是一个由一堆 XML 文件组成的 zip 文件，因此该进程需要一个有效的 ZIP 标头，但它没有找到它并省略了zipfile.BadZipfile: Truncated file header

看起来很奇怪的是您在调用堆栈中的错误消息，我本来希望看到print_to_excel 和load_workbook。

无论如何，既然您确认问题确实出在 XLSX 处理中，您也可以

通过tempfile 为每个进程生成一个新文件名
使用locking 确保一次只运行一个进程print_to_excel

【讨论】：