Pillow/Tests/threaded_save.py

59 lines
1.0 KiB
Python
Raw Normal View History

from __future__ import print_function
from PIL import Image
import io
2014-08-28 15:44:19 +04:00
import queue
import sys
import threading
import time
2015-04-24 11:24:52 +03:00
test_format = sys.argv[1] if len(sys.argv) > 1 else "PNG"
im = Image.open("Tests/images/hopper.ppm")
im.load()
queue = queue.Queue()
result = []
2014-08-28 15:44:19 +04:00
class Worker(threading.Thread):
def run(self):
2014-01-08 06:21:24 +04:00
while True:
im = queue.get()
if im is None:
queue.task_done()
sys.stdout.write("x")
break
f = io.BytesIO()
2015-04-24 11:24:52 +03:00
im.save(f, test_format, optimize=1)
data = f.getvalue()
result.append(len(data))
im = Image.open(io.BytesIO(data))
im.load()
sys.stdout.write(".")
queue.task_done()
2018-03-03 12:54:00 +03:00
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)