いわゆるスレッドセーフとは、あるリソースへのアクセスが、マルチスレッド状態でもシングルスレッド状態でも同じ結果が得られ、結果がスレッドスケジューリングなどの要因の影響を受けないことをいいます。

例えば、次のコードでは、関数 operate_resource() の中で g_list に新しい要素を追加し、新しい要素の値は g_list の最後の要素の値に 1 を加えたものになります。

g_list = []
def operate_resource():
        global g_list
        ele_num = len(g_list)
        if ele_num == 0:
                g_list.append(1)
        else:
                last_val = g_list[ele_num-1]
                new_last_val = last_val + 1
                g_list.append(new_last_val)

単一のスレッドで operate_resource() を繰り返し呼び出すと、g_list リストは絶えず増え、値は順に 1 ずつ増えます。しかし、2 つのスレッドを同時に起動し、それぞれのスレッドが operate_resource() をループで呼び出す場合、得られる g_list の値が毎回 1 ずつ増えることは保証されません。

上の例では、この 2 つのスレッドをそれぞれ A と B とします。現在、g_list は空で、A が 4 行目まで実行されると、ele_num=0 が得られるため、次の文は判定を通過して 6 行目が実行されます。このときスレッド切り替えが発生し、A スレッドがスケジューリングされて B がアクティブになったと仮定します。スレッド B も 4 行目のコードまで実行され、同様に B も ele_num=0 を得るため、6 行目のコードも実行されます。このようにして最後に得られる g_list は [1,1] になり、[1,2] ではなく、これは予期した結果ではありません。

この問題を完全に示すには、次のコードを使用できます。

import sys, time                        # 時間ライブラリを導入
if sys.version_info.major == 2:         # Python 2
    import thread
else:                                   # Python 3
    import _thread as thread
g_list = []
def operate_resource():                 # リソースを操作
    global g_list
    ele_num = len(g_list)
    if ele_num == 0:
        g_list.append(1)
    else:
        last_val = g_list[ele_num-1]
        new_last_val = last_val + 1
        g_list.append(new_last_val)
def thread_entry(id, round):             # スレッドエントリ関数
    while round > 0:
        operate_resource()
        round = round - 1
    print("Thread %d Finished" % id)
def start_threads():                     # スレッドを起動
    global g_list
    t1 = thread.start_new_thread(thread_entry, (1, 10000))
    t2 = thread.start_new_thread(thread_entry, (2, 10000))
    time.sleep(10)
    print("Check the Result")
    loc = 0             # 開始位置、それは順に g_list[loc] = (loc + 1) を満たすかどうかをチェック
    while loc < 20000:
        if g_list[loc] != (loc + 1):            # この値は予期したものと異なる
            print("Error: at %d" % (loc+1))
            break
        loc = loc + 1
if __name__=='__main__':
    start_threads()

実行結果は次の通りです：

$ python threadSafe1.py      # スクリプトを実行
Thread 2 Finished
Thread 1 Finished
Check the Result
Error: at 6488
$ python threadSafe1.py      # もう一度実行
Thread 2 Finished
Thread 1 Finished
Check the Result
Error: at 3666

最後に得られる結果は予期したものではないことがわかります。もちろん、このエラーデータには多少のランダム性があり、つまり、1 番目の要素でエラーが発生することもあれば、100 番目の要素でエラーが発生することもあります。例えば上の例では、2 回実行して、1 回は 6488 番目の要素でエラーが発生し、もう 1 回は 3666 番目の要素でエラーが発生しました。読者がこのプログラムを実行するときには、異なるエラー位置が得られる可能性もあります。

このエラーの原因は、ある操作が原子性を保証すべきであるにもかかわらず、保証されていないことです。上の例では、operate_resource() は原子性を保証する必要があり、つまり、operate_resource() の実行中に使用するリソースは、他のスレッドによって再利用されないようにする必要があります。つまり、スレッド A が g_list を操作している間、他のスレッドは g_list を使用できません。

解決策はロックを使用することで、基本的な使い方は次の通りです：

1. ロック A を取得し、他のスレッドが特定のデータを使用するのを阻止します。

2. 共有オブジェクトを使用します。

3. ロック A を解放し、他のスレッドが特定のデータを使用できるようにします。

これにより、任意の時点で、上の 2 番目のステップを実行しているのは 1 つのスレッドだけになります。

この方法と同期の違いは、1 つのロックだけを使用するため、あるスレッド A が連続して複数回ロックを取得する可能性があることです。一方、同期は順番に交互に実行されるため、あるスレッドが連続して複数回実行機会を得ることはあまりありません。

現在、上のコードを修正し、データにアクセスする前にロックをかけ、アクセスが終了したらロックを解放します。コードは次の通りです：

import sys, time
if sys.version_info.major == 2:
    import thread
else:
    import _thread as thread
mutex_lock = thread.allocate_lock()
g_list = []
def operate_resource():
    global g_list
    ele_num = len(g_list)
    if ele_num == 0:
        g_list.append(1)
    else:
        last_val = g_list[ele_num-1]
        new_last_val = last_val + 1
        g_list.append(new_last_val)
def thread_entry(id, round):··         # スレッドエントリ関数
    while round > 0:                   # round 回ループ
        mutex_lock.acquire()
        operate_resource()
        mutex_lock.release()
        round = round - 1
    print("Thread %d Finished" % id)
def start_threads():                   # スレッドを起動
    global g_list
    t1 = thread.start_new_thread(thread_entry, (1, 10000))
    t2 = thread.start_new_thread(thread_entry, (2, 10000))
    time.sleep(10)
    print("Check the Result")
    loc = 0
    while loc < 20000:
        if g_list[loc] != (loc + 1):
            print("Error: at %d" % (loc+1))
            break
        loc = loc + 1
    print("All Passed")
if __name__=='__main__':
    start_threads()

再度プログラムを実行すると、エラーは発生しません。

$ python threadSafe2.py
Thread 1 Finished
Thread 2 Finished
Check the Result
All Passed