Pillow/Tests/threaded_save.py
Hugo 5e676ea0bd Merge remote-tracking branch 'upstream/master' into flake8
Conflicts:
	Tests/bench_cffi_access.py
	Tests/test_file_palm.py
	Tests/test_format_hsv.py
	Tests/test_image_putdata.py
	Tests/test_locale.py
	Tests/test_mode_i16.py
2014-09-23 16:35:20 +03:00

60 lines
1010 B
Python

from PIL import Image
import io
import queue
import sys
import threading
import time
try:
format = sys.argv[1]
except:
format = "PNG"
im = Image.open("Tests/images/hopper.ppm")
im.load()
queue = queue.Queue()
result = []
class Worker(threading.Thread):
def run(self):
while True:
im = queue.get()
if im is None:
queue.task_done()
sys.stdout.write("x")
break
f = io.BytesIO()
im.save(f, format, optimize=1)
data = f.getvalue()
result.append(len(data))
im = Image.open(io.BytesIO(data))
im.load()
sys.stdout.write(".")
queue.task_done()
t0 = time.time()
threads = 20
jobs = 100
for i in range(threads):
w = Worker()
w.start()
for i in range(jobs):
queue.put(im)
for i in range(threads):
queue.put(None)
queue.join()
print()
print(time.time() - t0)
print(len(result), sum(result))
print(result)