偏った言語信者の垂れ流し

2016-04-14

[][]DjangoのDeprecationWarningを確認する

Djangoフレームワークでは、APIが変更、廃止される場合には、次のバージョンですぐに変わってしまうのではなく、警告を上げた上で2つ先のバージョンで削除されることになっています。

Deprecation policy

Pythonの組み込み例外であるDeprecationWarningを継承したクラスがDjangoでは定義されており、必要に応じてこれを確認できます。

確認方法

Pythonの実行オプションとして -Wd のように指定するとDeprecationWarningを表示できるので、これを「manage.py test」や「manage.py check」の実行時に指定すればよいです。

試したバージョンはPython3.5, Django1.9.5です。

$ python -Wd manage.py test

実行結果:

$ python -Wd manage.py test
Creating test database for alias 'default'...
/home/tokibito/sandbox/django_dep/myproject/myapp/urls.py:5: RemovedInDjango110Warning: Support for string view arguments to url() is deprecated and will be removed in Django 1.10 (got index). Pass the callable instead.
  url('^$', 'index'),
/home/tokibito/sandbox/django_dep/myproject/myapp/urls.py:5: RemovedInDjango110Warning: django.conf.urls.patterns() is deprecated and will be removed in Django 1.10. Update your urlpatterns to be a list of django.conf.urls.url() instances instead.
  url('^$', 'index'),
/home/tokibito/sandbox/django_dep/myproject/myproject/urls.py:20: RemovedInDjango20Warning: Passing a 3-tuple to django.conf.urls.include() is deprecated. Pass a 2-tuple containing the list of patterns and app_name, and provide the namespace argument to include() instead.
  url(r'^admin/', include(admin.site.urls)),
.
----------------------------------------------------------------------
Ran 1 test in 0.021s

OK

RemovedInDjango110WarningとRemovedInDjango20Warningの両方が表示されますが、20が不要な場合は、-Wオプションに指定する文字列を変更すれば除外できます。

$ python -W::DeprecationWarning manage.py test

これはRemovedInDjango20WarningがPendingDeprecationWarningを継承しているからです。

Pythonの実行オプションについては、ドキュメントを参照してください。

110を表示せずに20のみを表示したい場合は、PendingDeprecationWarningを指定するとできます。

$ python -W::PendingDeprecationWarning manage.py test

アップグレード作業をどのように進めるか

今、使っているバージョンでのDeprecationWarningを解消してからアップグレードすると、コードを大きく変更しなくても1つ次のバージョンでなら動作します。

先にDjangoのバージョンを上げてしまうと、アプリケーションは「修正しないと動かない」です。

アプリケーションが「動いてる状態」で警告を解消するほうが、手間は少ないです。

そのため、DeprecationWarningを修正しつつ、1つずつバージョンを上げていくのが結果的には楽なのかもしれないなと思いました。

もちろんテストコードは必須です。

参考

2016-04-06

[][][]cx_Oracleをマルチスレッドで使う

PythonOracle Databaseに接続するには、cx_Oracleを使う。

マルチスレッド環境下でcx_Oracleを使う場合にハマったのでメモを残す。

Oracleクライアントライブラリは、マルチスレッドで使う際には、OCI_THREADEDというモードで使わないといけないらしい。

cx_OracleをOCI_THREADEDモードで使うには、connect関数のthreadedオプションにTrueを指定する。

>>> import cx_Oracle
>>> connection = cx_Oracle.connect(
...     "testuser",
...     "testpassword",
...     "(DESCRIPTION=(ADDRESS=(PROTOCOL=TCP)(HOST=oracledb.example.com) (PORT=1521))(CONNECT_DATA=(SID=ORCL)))",
...     threaded=True)

cx_Oracleのドキュメントにも説明があった。

The threaded argument is expected to be a boolean expression which indicates whether or not Oracle should use the mode OCI_THREADED to wrap accesses to connections with a mutex.

Doing so in single threaded applications imposes a performance penalty of about 10-15% which is why the default is False.

Module Interface — cx_Oracle 5.2.1 documentation

DjangoOracleを使っていて、マルチスレッドで動作させる場合だと、settingsのデータベース設定に指定すればよい。

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.oracle',
        'NAME': 'oracledb.example.com:1521/orcl',
        'USER': 'testuser',
        'PASSWORD': 'testpassword',
        'OPTIONS': {
            'threaded': True,
        }
    }
}

この指定をしない状態で、gunicornのThreadWorkerで動かしていた際に、低頻度でワーカープロセスがログを出力せずに死ぬ、という現象が発生していた。

シングルスレッドの場合は、threaded=Falseのほうがパフォーマンスが良いそうなので、使いどころに気をつけたい。

参考

2016-03-28

[]concurrent.futures.ThreadPoolExecutorを使ってみる

Python 3.2から追加されたconcurrent.futuresモジュール

ThreadPoolExecutorを使うと、スレッドプールで処理を実行できる。

コード

main.py

import time
import threading
from concurrent.futures import ThreadPoolExecutor


def spam():
    for i in range(3):
        time.sleep(1)
        print("thread: {}, value: {}".format(threading.get_ident(), i))


def main():
    tpool = ThreadPoolExecutor(max_workers=3)
    for i in range(6):
        print("Threads: {}".format(len(tpool._threads)))  # スレッド数を表示
        tpool.submit(spam)
    print("main thread exit.")

実行結果

(venv) tokibito@tokibito-MacBookAir:~/sandbox$ python -c "import main;main.main()"
Threads: 0
Threads: 1
Threads: 2
Threads: 3
Threads: 3
Threads: 3
main thread exit.
thread: 140208052463360, value: 0
thread: 140208044070656, value: 0
thread: 140208035677952, value: 0
thread: 140208052463360, value: 1
thread: 140208044070656, value: 1
thread: 140208035677952, value: 1
thread: 140208052463360, value: 2
thread: 140208044070656, value: 2
thread: 140208035677952, value: 2
thread: 140208052463360, value: 0
thread: 140208044070656, value: 0
thread: 140208035677952, value: 0
thread: 140208044070656, value: 1
thread: 140208052463360, value: 1
thread: 140208035677952, value: 1
thread: 140208044070656, value: 2
thread: 140208052463360, value: 2
thread: 140208035677952, value: 2
  • スレッド数は必要になったときに増える
  • max_workersを超えるタスクを投げた場合は、キューに入れられて、ワーカースレッドが暇になったら次が実行される

参考

2016-03-27

[][]スレッドが終了するタイミングについて検証

Pythonスレッドをしばらく使っていなくて、どういう動きするのだったか忘れてたので、検証したメモを残す。

試した環境は、Ubuntu 14.04、Python 3.5.1。

検証に使ったコード

main.py:

import threading
import os
import sys
import time
import signal


def spam():
    for i in range(5):
        time.sleep(1)
        print("value: {}".format(i))


def main_1():
    t = threading.Thread(target=spam)
    t.start()
    print("Thread: {} started.".format(t.ident))
    sys.exit()


def main_2():
    t = threading.Thread(target=spam)
    t.start()
    print("Thread: {} started.".format(t.ident))
    t.join()  # スレッドの処理が終わるのを待つ
    print("Thread end.")
    sys.exit()


def main_3():
    t = threading.Thread(target=spam)
    t.start()
    print("Thread: {} started.".format(t.ident))
    # 自身のプロセスにSIGKILLを送って強制終了
    os.kill(os.getpid(), signal.SIGKILL)

実行結果

pythonの-cオプションを使って、各検証の関数を実行。

(venv) tokibito@tokibito-MacBookAir:~/sandbox$ python -c "import main;main.main_1()"
Thread: 140258192140032 started.
value: 0
value: 1
value: 2
value: 3
value: 4
(venv) tokibito@tokibito-MacBookAir:~/sandbox$ python -c "import main;main.main_2()"
Thread: 139913801033472 started.
value: 0
value: 1
value: 2
value: 3
value: 4
Thread end.
(venv) tokibito@tokibito-MacBookAir:~/sandbox$ python -c "import main;main.main_3()"
Thread: 140079644133120 started.
強制終了

参考

2016-03-18

[]funkload-friendlyというモジュールを作りました

FunkLoadPythonで作られたWebアプリケーション向けの負荷試験ツールです。

レポートがわかりやすいので、負荷試験の際によく利用してます。

便利ではあるのですが、いくつか使いづらい点があったので、funkload-friendlyというラッパーモジュールを作りました。

GitHub - tokibito/funkload-friendly: Friendly wrapper of FunkLoad.

Python 2.7のみ対応です(FunkLoadがPython3に未対応)

FunkLoadとの違い

大雑把ですが以下、funkload-friendlyで便利にしたところです。

なるべくFunkLoadの使用感を崩さないようにしてます。

インストール

virtualenvでどうぞ。

FunkLoadはバージョン1.17.1だとpipでインストールできないので、easy_installでインストールします。

funkload-friendlyはpipでインストールできます。

(venv)$ easy_install funkload
(venv)$ pip install funkload-friendly

使い方

ドキュメントは英語ですがあります。

funkload-friendly — funkload-friendly 0.2 documentation

http://localhost:8000/ で動いているDjangoアプリケーションに向けてテストするとします。

設定ファイルは以下のような感じに書けます。FunkLoadとの差は、指定できる項目が少し増えてるぐらい。

funkload.conf:

[main]
title = My Project Loadtest
description = load testing for myproject
url = http://localhost:8000

[ftest]
log_to = console file
log_directory = ./
result_directory = ./
sleep_time_min = 0
sleep_time_max = 0

[bench]
cycles = 1:3:10
duration = 10
startup_delay = 0.01
sleep_time = 0.01
cycle_time = 1
sleep_time_min = 0
sleep_time_max = 0.5
log_to = console file
log_directory = ./
result_directory = ./

負荷試験のコードは、funkload_friendly.test.TestCaseを継承して書きます。

ログインAPI呼び出しの検証は、FunkLoadをそのまま使うより簡単に書けます。

loadtest.py:

from funkload_friendly.test import TestCase, description
from funkload_friendly.datatypes import JSONData


class MainTest(TestCase):
    @description("Load top_page")
    def test_top_page(self):
        self.get(self.site_url + "/")


class LoginTest(TestCase):
    def setUp(self):
        # login
        self.get(self.site_url + "/login/")
        self.post(self.site_url + "/login/",
            params=[
                ['username', 'spam'],
                ['password', 'P@ssw0rd'],
                ['csrfmiddlewaretoken', self.cookie['csrftoken']],
            ]
        )

    @description("Load secret_page with login")
    def test_secret_page(self):
        response = self.get(self.site_url + "/secret_page")
        self.assertEqual(response.code, 200)


class APITest(TestCase):
    @description("Load REST API")
    def test_calculate_add(self):
        response = self.post(self.site_url + "/calculate/add/",
            params=JSONData({
                'value1': 100,
                'value2': 50,
            })
        )
        self.assertEqual(response.code, 200)
        self.assertEqual(response.data['result'], 150)

試験の実行は、FunkLoadのコマンドをそのまま使います。

$ fl-run-bench --config=funkload.conf loadtest MainTest.test_top_page

プロジェクトはcookiecutterを使って作成することもできます。

GitHub - tokibito/cookiecutter-funkload-friendly: Cookiecutter template for a funkload-friendly project.

サンプルコード

funkload-friendlyを使った負荷試験のコードと、試験対象のDjangoのプロジェクトをGithubに置いています。

GitHub - tokibito/funkload-friendly-example: Example code for funkload-friendly