boto3のマルチパート転送時のハンドルリーク

Question

質問

boto3を使用してプログラムを常駐させて、s3へのファイルアップロードを行っているのですが、gcでもハンドルが回収されず右肩上りになっています。
以下の設定をupload_fileの引数に指定するでリーク自体は収まるのでマルチパート転送時固有の問題みたいです。

TransferConfig(use_threads=False)

そこで質問です。
・pythonでハンドルリークを検出する方法はあるのでしょうか
・boto3でオブジェクトを開放するよう外部から呼び出せる関数はあるのでしょうか

環境

・windows10
・python 3.6(64bit)
・boto3 1.9.23
・botocore 1.12.23

根拠

Windowsのパフォーマンスカウンターで\Process(python)\Handle Countで見ております。

検証コード

class s3Session:
    def __init__(self, access_key, secret_access_key, region_name=None):
        self._session = Session(aws_access_key_id=access_key,
                               aws_secret_access_key=secret_access_key,
                               region_name=region_name)

    def __enter__(self):
        return self

    def __exit__(self, type, value, traceback):
        return True

    @property
    def client(self):
        return self._session.resource('s3').meta.client

def s3_upload_loop(s3Session, writeByte, bucket):
    while True:
        try:
            # ダミーのファイル出力
            with tempfile.NamedTemporaryFile(delete=False) as tempFile:
                tempFile.write(numpy.random.bytes(int(writeByte)))
                tempName = tempFile.name

            s3Session.client.upload_file(tempName, bucket, os.path.basename(tempName))
            gc.collect()
        except botocore.exceptions.EndpointConnectionError:
            print("Network Error: Please Check your Internet Connection")
        except Exception as e:
            print(str(e))
        finally:
            os.remove(tempName)


if __name__ == '__main__':
    bucket = 'バケット名'

    with s3Session('アクセスキー', 'シークレットキー') as sesstion :
        s3_upload_loop(sesstion, 10*1024*1024, bucket)

Comments

[YouheiSakurai]

どんなハンドルがリークしているか「http://replication.hatenablog.com/entry/2015/06/16/093000」で調べてもらえませんか？

[TomeSq]

ProcessExplorerで同じハンドル数にしかならなかったので、同じsysinternalsのHandleツールで確認するとどうもSemaphoreハンドルの開放が漏れているようです。1ループ目：143、2ループ目：148となっていました。

Answer 1

転送する際、以下のようにTransferManagerを経由でS3Transferを作成して使いまわして上げればリークしないことがわかりました。

def s3_upload_loop(access_key, secret_access_key, writeByte, bucket):

    config = TransferConfig(use_threads = True)
    client = boto3.session.Session(access_key, secret_access_key).resource('s3').meta.client
    with TransferManager(client, config) as manager :
        with S3Transfer(manager = manager) as transfer:
            while True:
                try:
                    # ダミーのファイル出力
                    with tempfile.NamedTemporaryFile(delete=False) as tempFile:
                        tempFile.write(numpy.random.bytes(int(writeByte)))
                        tempName = tempFile.name

                    transfer.upload_file(tempName, bucket, os.path.basename(tempName))

                    gc.collect()
                except botocore.exceptions.EndpointConnectionError:
                    print("Network Error: Please Check your Internet Connection")
                except Exception as e:
                    print(str(e))
                finally:
                    os.remove(tempName)

違いを調べたところupload_file内でThreadPoolExecutorを毎回作成しているのが原因でした。TransferManager経由だと作成済みのThreadPoolExecutorを使いまわしているのでリークしないようです。

単純に以下の処理だけでもリークするのでThreadPoolExecutorの終了処理？がうまく行っていないように見受けられます。

def task():
    pass

if __name__ == '__main__':
    while True:
        with ThreadPoolExecutor(max_workers=2, thread_name_prefix="thread") as executor:
            executor.submit(task)

ただ、使い方がわるいのか、そもそもThreadPoolExecutorの問題なのかわかりませんがとりあえず原因と回避方法はわかりました。

Answer 2

以下にPythonでプロセス情報を取得する必要がありますので、
これを実装してカウントすれば良いと思います。

http://www.denzow.me/entry/2017/09/13/235856

Answer 3

以下のコードで再現すればGitHubのIssueへレポートするのがいいんじゃないでしょうか。

from io import BytesIO
from os import environ
from boto3 import Session


creds = dict(aws_access_key_id=environ.get("AWS_ACCESS_KEY_ID"),
             aws_secret_access_key=environ.get("AWS_SECRET_ACCESS_KEY"),
             region_name=environ.get("AWS_DEFAULT_REGION "))


with Session(**creds) as sess, BytesIO("DUMMY") as fp:
    client = sess.resource("s3").meta.client
    for i in range(10):
        fp.seek(0)
        client.upload_fileobj(fp, "my-test-bucket-2018-11-06", "obj-%d" % i)

ちなみにWindows環境でハンドルまわりをPythonで扱うためのコードは以下みたいになります。ご興味があれば。

'''
Created on 2017/11/07

@author: sakurai
'''
from ctypes import byref
from ctypes import cast
from ctypes import create_string_buffer
from ctypes import create_unicode_buffer
from ctypes import POINTER
from ctypes import Structure
from ctypes import windll
from ctypes import WinDLL
from ctypes import WinError
from ctypes import wintypes
from ctypes.wintypes import LPVOID as PVOID
from os.path import isfile


NTDLL = WinDLL("ntdll")
KERNEL32 = windll.kernel32
NTSTATUS = wintypes.LONG
SYSTEM_INFORMATION_CLASS = wintypes.ULONG


####
# C structures
####
class SYSTEM_HANDLE(Structure):
    """https://stackoverflow.com/a/5163277"""
    _fields_ = (("dwProcessId", wintypes.DWORD),
                ("bObjectType", wintypes.BYTE),
                ("bFlags", wintypes.BYTE),
                ("wValue", wintypes.WORD),
                ("pAddress", PVOID),
                ("GrantedAccess", wintypes.DWORD))


class SYSTEM_HANDLE_INFORMATION(Structure):
    _fields_ = (("dwCount", wintypes.DWORD),
                ("Handles", SYSTEM_HANDLE * 1))


####
# C error checks
####
def assertZero(result, _, args):
        if result != 0:
            raise WinError(NTDLL.RtlNtStatusToDosError(result))
        return args


def assertNotZero(result, _, args):
    if result == 0:
        raise WinError()
    return args


def assertTrue(result, _, args):
    if not result:
        raise WinError()
    return args


####
# C functions
####
NtQuerySystemInformation = NTDLL.NtQuerySystemInformation
NtQuerySystemInformation.restype = NTSTATUS
NtQuerySystemInformation.argtypes = (SYSTEM_INFORMATION_CLASS,
                                     PVOID,
                                     wintypes.ULONG,
                                     wintypes.PULONG)
NtQuerySystemInformation.errcheck = assertZero

GetFinalPathNameByHandle = KERNEL32.GetFinalPathNameByHandleW
GetFinalPathNameByHandle.restype = wintypes.DWORD
GetFinalPathNameByHandle.argtypes = (wintypes.HANDLE,
                                     wintypes.LPWSTR,
                                     wintypes.DWORD,
                                     wintypes.DWORD)
GetFinalPathNameByHandle.errcheck = assertNotZero

CloseHandle = KERNEL32.CloseHandle
CloseHandle.restype = wintypes.BOOL
CloseHandle.argtypes = (wintypes.HANDLE, )
CloseHandle.errcheck = assertTrue


####
# Wrappers
####
def system_handles():
    dwSize = wintypes.DWORD(0)
    pInfo = create_string_buffer(dwSize.value)
    while True:
        try:
            NtQuerySystemInformation(0x10,  # SystemHandleInformation
                                     pInfo, dwSize, byref(dwSize))
            break
        except OSError as e:
            if e.errno == 13:
                pInfo = create_string_buffer(dwSize.value)
            else:
                raise e

    info = cast(pInfo, POINTER(SYSTEM_HANDLE_INFORMATION)).contents
    yield from cast(info.Handles,
                    POINTER(SYSTEM_HANDLE * info.dwCount)).contents


def get_path(handle: wintypes.HANDLE):
    szPath = create_unicode_buffer(wintypes.MAX_PATH)
    GetFinalPathNameByHandle(handle, szPath, wintypes.MAX_PATH, 0)
    return szPath.value


def open_files(pid: int):
    for handle in system_handles():
        if handle.dwProcessId == pid:
            try:
                fname = get_path(handle.wValue)
                if isfile(fname):
                    yield (get_path(handle.wValue), handle.wValue)
            except OSError:
                pass