1*9acde482SJonathan Wakely# Copyright (c) 2016-2023 Intel Corporation 251c0b2f7Stbbdev# 351c0b2f7Stbbdev# Licensed under the Apache License, Version 2.0 (the "License"); 451c0b2f7Stbbdev# you may not use this file except in compliance with the License. 551c0b2f7Stbbdev# You may obtain a copy of the License at 651c0b2f7Stbbdev# 751c0b2f7Stbbdev# http://www.apache.org/licenses/LICENSE-2.0 851c0b2f7Stbbdev# 951c0b2f7Stbbdev# Unless required by applicable law or agreed to in writing, software 1051c0b2f7Stbbdev# distributed under the License is distributed on an "AS IS" BASIS, 1151c0b2f7Stbbdev# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 1251c0b2f7Stbbdev# See the License for the specific language governing permissions and 1351c0b2f7Stbbdev# limitations under the License. 1451c0b2f7Stbbdev 1551c0b2f7Stbbdev# Based on the software developed by: 1651c0b2f7Stbbdev# Copyright (c) 2008,2016 david decotigny (Pool of threads) 1751c0b2f7Stbbdev# Copyright (c) 2006-2008, R Oudkerk (multiprocessing.Pool) 1851c0b2f7Stbbdev# All rights reserved. 1951c0b2f7Stbbdev# 2051c0b2f7Stbbdev# Redistribution and use in source and binary forms, with or without 2151c0b2f7Stbbdev# modification, are permitted provided that the following conditions 2251c0b2f7Stbbdev# are met: 2351c0b2f7Stbbdev# 2451c0b2f7Stbbdev# 1. Redistributions of source code must retain the above copyright 2551c0b2f7Stbbdev# notice, this list of conditions and the following disclaimer. 2651c0b2f7Stbbdev# 2. Redistributions in binary form must reproduce the above copyright 2751c0b2f7Stbbdev# notice, this list of conditions and the following disclaimer in the 2851c0b2f7Stbbdev# documentation and/or other materials provided with the distribution. 2951c0b2f7Stbbdev# 3. Neither the name of author nor the names of any contributors may be 3051c0b2f7Stbbdev# used to endorse or promote products derived from this software 3151c0b2f7Stbbdev# without specific prior written permission. 3251c0b2f7Stbbdev# 3351c0b2f7Stbbdev# THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS "AS IS" AND 3451c0b2f7Stbbdev# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 3551c0b2f7Stbbdev# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 3651c0b2f7Stbbdev# ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE 3751c0b2f7Stbbdev# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL 3851c0b2f7Stbbdev# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS 3951c0b2f7Stbbdev# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) 4051c0b2f7Stbbdev# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT 4151c0b2f7Stbbdev# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY 4251c0b2f7Stbbdev# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF 4351c0b2f7Stbbdev# SUCH DAMAGE. 4451c0b2f7Stbbdev# 4551c0b2f7Stbbdev 4651c0b2f7Stbbdevimport time 4751c0b2f7Stbbdevimport threading 4851c0b2f7Stbbdev 4951c0b2f7Stbbdevfrom .api import * 5051c0b2f7Stbbdevfrom .pool import * 5151c0b2f7Stbbdev 5251c0b2f7Stbbdev 5351c0b2f7Stbbdevdef test(arg=None): 5451c0b2f7Stbbdev if arg == "-v": 5551c0b2f7Stbbdev def say(*x): 5651c0b2f7Stbbdev print(*x) 5751c0b2f7Stbbdev else: 5851c0b2f7Stbbdev def say(*x): 5951c0b2f7Stbbdev pass 6051c0b2f7Stbbdev say("Start Pool testing") 6151c0b2f7Stbbdev print("oneTBB version is %s" % runtime_version()) 6251c0b2f7Stbbdev print("oneTBB interface version is %s" % runtime_interface_version()) 6351c0b2f7Stbbdev 6451c0b2f7Stbbdev get_tid = lambda: threading.current_thread().ident 6551c0b2f7Stbbdev 6651c0b2f7Stbbdev assert default_num_threads() == this_task_arena_max_concurrency() 6751c0b2f7Stbbdev 6851c0b2f7Stbbdev def return42(): 6951c0b2f7Stbbdev return 42 7051c0b2f7Stbbdev 7151c0b2f7Stbbdev def f(x): 7251c0b2f7Stbbdev return x * x 7351c0b2f7Stbbdev 7451c0b2f7Stbbdev def work(mseconds): 7551c0b2f7Stbbdev res = str(mseconds) 7651c0b2f7Stbbdev if mseconds < 0: 7751c0b2f7Stbbdev mseconds = -mseconds 7851c0b2f7Stbbdev say("[%d] Start to work for %fms..." % (get_tid(), mseconds*10)) 7951c0b2f7Stbbdev time.sleep(mseconds/100.) 8051c0b2f7Stbbdev say("[%d] Work done (%fms)." % (get_tid(), mseconds*10)) 8151c0b2f7Stbbdev return res 8251c0b2f7Stbbdev 832eccd5f9SIlya Isaev # special flag to to be set by thread calling async work 842eccd5f9SIlya Isaev spin_flag = None 852eccd5f9SIlya Isaev def timeout_work(param): 862eccd5f9SIlya Isaev say("[%d] Spin wait work start..." % get_tid()) 872eccd5f9SIlya Isaev while spin_flag: 882eccd5f9SIlya Isaev time.sleep(0.0001) # yield equivalent 892eccd5f9SIlya Isaev say("[%d] Work done." % get_tid()) 902eccd5f9SIlya Isaev return str(param) if param != None else None 912eccd5f9SIlya Isaev 922eccd5f9SIlya Isaev def prepare_timeout_exception(): 932eccd5f9SIlya Isaev nonlocal spin_flag 942eccd5f9SIlya Isaev spin_flag = True # lock threads in timeout_work 952eccd5f9SIlya Isaev 962eccd5f9SIlya Isaev def check_timeout_exception(pool_object, func): 972eccd5f9SIlya Isaev nonlocal spin_flag 982eccd5f9SIlya Isaev try: 992eccd5f9SIlya Isaev func(pool_object) 1002eccd5f9SIlya Isaev except TimeoutError: 1012eccd5f9SIlya Isaev say("Good. Got expected timeout exception.") 1022eccd5f9SIlya Isaev else: 1032eccd5f9SIlya Isaev assert False, "Expected exception !" 1042eccd5f9SIlya Isaev spin_flag = False # unlock threads in timeout_work 1052eccd5f9SIlya Isaev 10651c0b2f7Stbbdev ### Test copy/pasted from multiprocessing 10751c0b2f7Stbbdev pool = Pool(4) # start worker threads 10851c0b2f7Stbbdev 10951c0b2f7Stbbdev # edge cases 11051c0b2f7Stbbdev assert pool.map(return42, []) == [] 11151c0b2f7Stbbdev assert pool.apply_async(return42, []).get() == 42 11251c0b2f7Stbbdev assert pool.apply(return42, []) == 42 11351c0b2f7Stbbdev assert list(pool.imap(return42, iter([]))) == [] 11451c0b2f7Stbbdev assert list(pool.imap_unordered(return42, iter([]))) == [] 11551c0b2f7Stbbdev assert pool.map_async(return42, []).get() == [] 11651c0b2f7Stbbdev assert list(pool.imap_async(return42, iter([])).get()) == [] 11751c0b2f7Stbbdev assert list(pool.imap_unordered_async(return42, iter([])).get()) == [] 11851c0b2f7Stbbdev 11951c0b2f7Stbbdev # basic tests 12051c0b2f7Stbbdev result = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously 12151c0b2f7Stbbdev assert result.get(timeout=1) == 100 # ... unless slow computer 12251c0b2f7Stbbdev assert list(pool.map(f, range(10))) == list(map(f, range(10))) 12351c0b2f7Stbbdev it = pool.imap(f, range(10)) 12451c0b2f7Stbbdev assert next(it) == 0 12551c0b2f7Stbbdev assert next(it) == 1 12651c0b2f7Stbbdev assert next(it) == 4 12751c0b2f7Stbbdev 12851c0b2f7Stbbdev # Test apply_sync exceptions 1292eccd5f9SIlya Isaev prepare_timeout_exception() 1302eccd5f9SIlya Isaev result = pool.apply_async(timeout_work, (None,)) 1312eccd5f9SIlya Isaev check_timeout_exception(result, lambda result : say(result.get(timeout=1))) 13251c0b2f7Stbbdev assert result.get() is None # sleep() returns None 13351c0b2f7Stbbdev 13451c0b2f7Stbbdev def cb(s): 13551c0b2f7Stbbdev say("Result ready: %s" % s) 13651c0b2f7Stbbdev 13751c0b2f7Stbbdev # Test imap() 13851c0b2f7Stbbdev assert list(pool.imap(work, range(10, 3, -1), chunksize=4)) == list(map( 13951c0b2f7Stbbdev str, range(10, 3, -1))) 14051c0b2f7Stbbdev 14151c0b2f7Stbbdev # Test imap_unordered() 14251c0b2f7Stbbdev assert sorted(pool.imap_unordered(work, range(10, 3, -1))) == sorted(map( 14351c0b2f7Stbbdev str, range(10, 3, -1))) 14451c0b2f7Stbbdev 14551c0b2f7Stbbdev # Test map_async() 1462eccd5f9SIlya Isaev prepare_timeout_exception() 1472eccd5f9SIlya Isaev result = pool.map_async(timeout_work, range(10), callback=cb) 1482eccd5f9SIlya Isaev check_timeout_exception(result, lambda result : result.get(timeout=0.01)) 14951c0b2f7Stbbdev say(result.get()) 15051c0b2f7Stbbdev 15151c0b2f7Stbbdev # Test imap_async() 1522eccd5f9SIlya Isaev prepare_timeout_exception() 1532eccd5f9SIlya Isaev result = pool.imap_async(timeout_work, range(3, 10), callback=cb) 1542eccd5f9SIlya Isaev check_timeout_exception(result, lambda result : result.get(timeout=0.01)) 15551c0b2f7Stbbdev for i in result.get(): 15651c0b2f7Stbbdev say("Item:", i) 15751c0b2f7Stbbdev say("### Loop again:") 15851c0b2f7Stbbdev for i in result.get(): 15951c0b2f7Stbbdev say("Item2:", i) 16051c0b2f7Stbbdev 16151c0b2f7Stbbdev # Test imap_unordered_async() 1622eccd5f9SIlya Isaev prepare_timeout_exception() 1632eccd5f9SIlya Isaev result = pool.imap_unordered_async(timeout_work, range(10, 3, -1), callback=cb) 1642eccd5f9SIlya Isaev check_timeout_exception(result, lambda result : result.get(timeout=0.01)) 16551c0b2f7Stbbdev for i in result.get(): 16651c0b2f7Stbbdev say("Item1:", i) 16751c0b2f7Stbbdev for i in result.get(): 16851c0b2f7Stbbdev say("Item2:", i) 16951c0b2f7Stbbdev r = result.get() 17051c0b2f7Stbbdev for i in r: 17151c0b2f7Stbbdev say("Item3:", i) 17251c0b2f7Stbbdev for i in r: 17351c0b2f7Stbbdev say("Item4:", i) 17451c0b2f7Stbbdev for i in r: 17551c0b2f7Stbbdev say("Item5:", i) 17651c0b2f7Stbbdev 17751c0b2f7Stbbdev # 17851c0b2f7Stbbdev # The case for the exceptions 17951c0b2f7Stbbdev # 18051c0b2f7Stbbdev 18151c0b2f7Stbbdev # Exceptions in imap_unordered_async() 18251c0b2f7Stbbdev result = pool.imap_unordered_async(work, range(2, -10, -1), callback=cb) 18351c0b2f7Stbbdev time.sleep(3) 18451c0b2f7Stbbdev try: 18551c0b2f7Stbbdev for i in result.get(): 18651c0b2f7Stbbdev say("Got item:", i) 18751c0b2f7Stbbdev except (IOError, ValueError): 18851c0b2f7Stbbdev say("Good. Got expected exception") 18951c0b2f7Stbbdev 19051c0b2f7Stbbdev # Exceptions in imap_async() 19151c0b2f7Stbbdev result = pool.imap_async(work, range(2, -10, -1), callback=cb) 19251c0b2f7Stbbdev time.sleep(3) 19351c0b2f7Stbbdev try: 19451c0b2f7Stbbdev for i in result.get(): 19551c0b2f7Stbbdev say("Got item:", i) 19651c0b2f7Stbbdev except (IOError, ValueError): 19751c0b2f7Stbbdev say("Good. Got expected exception") 19851c0b2f7Stbbdev 19951c0b2f7Stbbdev # Stop the test: need to stop the pool !!! 20051c0b2f7Stbbdev pool.terminate() 20151c0b2f7Stbbdev pool.join() 20251c0b2f7Stbbdev 20351c0b2f7Stbbdevif __name__ == "__main__": 20451c0b2f7Stbbdev test() 205