| Server IP : 109.234.164.53 / Your IP : 216.73.216.110 Web Server : Apache System : Linux cervelle.o2switch.net 4.18.0-553.32.1.lve.el8.x86_64 #1 SMP Thu Dec 19 13:14:03 UTC 2024 x86_64 User : computer3 ( 1098) PHP Version : 7.1.33 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /proc/thread-self/root/opt/alt/python311/lib64/python3.11/test/ |
Upload File : |
import multiprocessing
import random
import sys
import time
def fill_queue(queue, code):
queue.put(code)
def drain_queue(queue, code):
if code != queue.get():
sys.exit(1)
def test_func():
code = random.randrange(0, 1000)
queue = multiprocessing.Queue()
fill_pool = multiprocessing.Process(
target=fill_queue,
args=(queue, code)
)
drain_pool = multiprocessing.Process(
target=drain_queue,
args=(queue, code)
)
drain_pool.start()
fill_pool.start()
fill_pool.join()
drain_pool.join()
def main():
test_pool = multiprocessing.Process(target=test_func)
test_pool.start()
test_pool.join()
sys.exit(test_pool.exitcode)
if __name__ == "__main__":
main()