في البداية كان
هذا المقال . ثم ظهر
تعليق عليها . نتيجة لذلك ، بحثت في قراءة العتاد ، وحفرت في debag وكنت قادراً على تحسين الكود من الجزء الأول من هذه القصة. أقترح المشي معي على طول النقاط الرئيسية.
أولاً ، أود أن أشكر
موغوست . بفضل تعليقه ، أعدت تعريف طريقة بيثون. لقد سبق أن سمعت أن هناك الكثير من الرجال غير الاقتصاديين بين الثوار (عند التعامل مع الذاكرة) ، والآن اتضح أنني انضممت إلى هذا الحزب بطريقة غير مرئية.
لذلك دعونا نبدأ. دعونا التكهن بشأن ما كانت الاختناقات بشكل عام.
مستمر إذا:
if isinstance(self.custom_handlers, property): if self.custom_handlers and e.__class__ in self.custom_handlers: if e.__class__ not in self.exclude:
وهذا ليس الحد. لذلك ، قمت بإزالة جزء من ifs ، وقمت بنقل شيء إلى __init__ ، أي إلى حيث سيتم استدعاء مرة واحدة. على وجه التحديد ، يجب استدعاء التحقق من الخاصية في الكود مرة واحدة ، لأن يتم تطبيق الديكور على الطريقة المعينة لها. وستظل ملكية الفئة ، على التوالي ، دون تغيير. لذلك ، ليست هناك حاجة للتحقق من الممتلكات باستمرار.
نقطة منفصلة هي إذا في. أظهر المحلل أن لكل من لديه دعوة منفصلة ، لذلك قررت أن أجمع كل معالجات في dict واحد. هذا يسمح بتجنب ifs بشكل عام ، بدلاً من استخدام ببساطة:
self.handlers.get(e.__class__, Exception)(e)
لذلك في برامج self.handlers لدينا إملاء ، والذي يحتوي افتراضيًا على دالة تثير الاستثناءات
الأخرى .
بالطبع ، المجمع يستحق اهتماما خاصا. هذه هي نفس الوظيفة التي تسمى في كل مرة يتم استدعاء الديكور. أي من الأفضل هنا تجنب عمليات الفحص غير الضرورية وجميع أنواع الأحمال إلى الحد الأقصى ، إن أمكن ، وإخراجهما في __init__ أو __call__. وإليك ما كان المجمع قبل:
def wrapper(self, *args, **kwargs): if self.custom_handlers: if isinstance(self.custom_handlers, property): self.custom_handlers = self.custom_handlers.__get__(self, self.__class__) if asyncio.iscoroutinefunction(self.func): return self._coroutine_exception_handler(*args, **kwargs) else: return self._sync_exception_handler(*args, **kwargs)
عدد الشيكات يمر عبر السقف. سيتم استدعاء كل هذا عند كل مكالمة إلى الديكور. لذلك أصبح المجمع مثل هذا:
def __call__(self, func): self.func = func if iscoroutinefunction(self.func): def wrapper(*args, **kwargs): return self._coroutine_exception_handler(*args, **kwargs) else: def wrapper(*args, **kwargs): return self._sync_exception_handler(*args, **kwargs) return wrapper
أذكر ، سيتم استدعاء __call__ مرة واحدة. داخل __call__ ، استنادًا إلى درجة عدم تزامن الوظيفة ، نرجع الدالة نفسها أو coroutine. وأريد أيضًا أن أشير إلى أن asyncio.iscoroutinefunction تجري مكالمة إضافية ، لذلك انتقلت إلى inspect.iscoroutinefunction. في الواقع ، المقاعد (cProfile) للتزامن والتفتيش:
ncalls tottime percall cumtime percall filename:lineno(function) 1 0.000 0.000 0.000 0.000 <string>:1(<module>) 1 0.000 0.000 0.000 0.000 coroutines.py:160(iscoroutinefunction) 1 0.000 0.000 0.000 0.000 inspect.py:158(isfunction) 1 0.000 0.000 0.000 0.000 inspect.py:179(iscoroutinefunction) 1 0.000 0.000 0.000 0.000 {built-in method builtins.exec} 1 0.000 0.000 0.000 0.000 {built-in method builtins.isinstance} 1 0.000 0.000 0.000 0.000 {method 'disable' of '_lsprof.Profiler' objects}
ncalls tottime percall cumtime percall filename:lineno(function) 1 0.000 0.000 0.000 0.000 <string>:1(<module>) 1 0.000 0.000 0.000 0.000 inspect.py:158(isfunction) 1 0.000 0.000 0.000 0.000 inspect.py:179(iscoroutinefunction) 1 0.000 0.000 0.000 0.000 {built-in method builtins.exec} 1 0.000 0.000 0.000 0.000 {built-in method builtins.isinstance} 1 0.000 0.000 0.000 0.000 {method 'disable' of '_lsprof.Profiler' objects}
كود كامل:
from inspect import iscoroutinefunction from asyncio import QueueEmpty, QueueFull from concurrent.futures import TimeoutError class ProcessException(object): __slots__ = ('func', 'handlers') def __init__(self, custom_handlers=None): self.func = None if isinstance(custom_handlers, property): custom_handlers = custom_handlers.__get__(self, self.__class__) def raise_exception(e: Exception): raise e exclude = { QueueEmpty: lambda e: None, QueueFull: lambda e: None, TimeoutError: lambda e: None } self.handlers = { **exclude, **(custom_handlers or {}), Exception: raise_exception } def __call__(self, func): self.func = func if iscoroutinefunction(self.func): def wrapper(*args, **kwargs): return self._coroutine_exception_handler(*args, **kwargs) else: def wrapper(*args, **kwargs): return self._sync_exception_handler(*args, **kwargs) return wrapper async def _coroutine_exception_handler(self, *args, **kwargs): try: return await self.func(*args, **kwargs) except Exception as e: return self.handlers.get(e.__class__, Exception)(e) def _sync_exception_handler(self, *args, **kwargs): try: return self.func(*args, **kwargs) except Exception as e: return self.handlers.get(e.__class__, Exception)(e)
وربما يكون المثال غير مكتمل بدون وقت. لذلك ، باستخدام المثال من التعليق أعلاه:
class MathWithTry(object): def divide(self, a, b): try: return a // b except ZeroDivisionError: return ' , '
ومثال من نص
المقال السابق (
تنبيه! نمرر
e إلى المثال من النص الموجود في lambda. لم يكن هذا هو الحال في المقالة السابقة وتمت إضافته فقط في الابتكارات):
class Math(object): @property def exception_handlers(self): return { ZeroDivisionError: lambda e: ' , ' } @ProcessException(exception_handlers) def divide(self, a, b): return a // b
وهنا النتائج بالنسبة لك:
timeit.timeit('math_with_try.divide(1, 0)', number=100000, setup='from __main__ import math_with_try') 0.05079065300014918 timeit.timeit('math_with_decorator.divide(1, 0)', number=100000, setup='from __main__ import math_with_decorator') 0.16211646200099494
في الختام ، أود أن أقول إن التحسين ، في رأيي ، عملية معقدة إلى حد ما ، وهنا من المهم عدم الابتعاد وعدم تحسين شيء على حساب قابلية القراءة. خلاف ذلك ، سيكون الخصم على التحسين أمراً بالغ الصعوبة.
شكرا لتعليقاتكم وإنني أتطلع إلى التعليقات على هذه المادة أيضا :)
ملاحظة: بفضل تعليقات مستخدمي Habr ، كان من الممكن تسريع أكثر ، وهذا ما حدث:
from inspect import iscoroutinefunction from asyncio import QueueEmpty, QueueFull from concurrent.futures import TimeoutError class ProcessException(object): __slots__ = ('handlers',) def __init__(self, custom_handlers=None): if isinstance(custom_handlers, property): custom_handlers = custom_handlers.__get__(self, self.__class__) raise_exception = ProcessException.raise_exception exclude = { QueueEmpty: lambda e: None, QueueFull: lambda e: None, TimeoutError: lambda e: None } self.handlers = { **exclude, **(custom_handlers or {}), Exception: raise_exception } def __call__(self, func): handlers = self.handlers if iscoroutinefunction(func): async def wrapper(*args, **kwargs): try: return await func(*args, **kwargs) except Exception as e: return handlers.get(e.__class__, handlers[Exception])(e) else: def wrapper(*args, **kwargs): try: return func(*args, **kwargs) except Exception as e: return handlers.get(e.__class__, handlers[Exception])(e) return wrapper @staticmethod def raise_exception(e: Exception): raise e
timeit.timeit('divide(1, 0)', number=100000, setup='from __main__ import divide') 0.13714907199755544
تسارعت بنسبة 0.03 في المتوسط. بفضل
كوستيانتين و
Yngvie .
PS تحديث! لقد قمت بتحسين الرمز بناءً على تعليقات من تعليقات
onegreyonewhite و
resetme . استبدال self.func مع فقط func وجعل self.handlers إلى متغير. تم تسريع التنفيذ ، خاصةً إذا كان هناك المزيد من التكرار لكل صفر. أقتبس timeit:
timeit.timeit('t.divide_with_decorator(1, 0)', number=1000000, setup='from __main__ import t') 1.1116105649998644
قبل هذا التحسين ، استغرق التنفيذ بنفس قيمة الرقم 1.24 في المتوسط.
ملحوظة: لقد قمت بتحسيني أكثر من خلال إدخال الدالة lift_exception من __init__ في @ staticthethod وأنا أتصل بها من خلال متغير لإزالة الوصول عبر نقطة. في الواقع ، أصبح متوسط وقت التنفيذ:
timeit.timeit('t.divide_with_decorator(1, 0)', number=1000000, setup='from __main__ import t') 1.0691639049982768
هذا هو للطريقة. وتسمى الوظائف بشكل أسرع (في المتوسط):
timeit.timeit('div(1, 0)', number=1000000, setup='from __main__ import div') 1.0463485610016505