استخدام غير متزامن لإنشاء برامج تشغيل الأجهزة غير المتزامنة على MicroPython v.1.12

أثناء دراسة إمكانيات MicroPython لأغراضها ، صادفت أحد تطبيقات مكتبة asyncio ، وبعد مراسلات قصيرة مع Piter Hinch ، مؤلف المكتبة ، أدركت أنني بحاجة إلى فهم المبادئ والمفاهيم الأساسية والأخطاء النموذجية بشكل أكبر لاستخدام طرق البرمجة غير المتزامنة. علاوة على ذلك ، قسم المبتدئين مخصص لي فقط.

هذا الدليل مخصص للمستخدمين ذوي مستويات مختلفة من الخبرة في المزامنة ، بما في ذلك قسم خاص للمبتدئين.

محتوى
0. مقدمة
0.1. ___ تثبيت uasyncio على جهاز فارغ (الأجهزة)
1. التخطيط لتنفيذ البرنامج المشترك
1.1
2. مكتبة uasyncio
2.1. ___ هيكل البرنامج: دورة معالجة الأحداث
2.2
2.2.1. ____ قائمة انتظار coroutines للمشاركة في التخطيط
2.2.2. ___ بدء رد اتصال وظيفة ( رد اتصال )
2.2.3. ___ الملاحظات: coroutines كطرق ذات صلة. قيم الإرجاع.
2.3. التأخير
3. التزامن وفئاته
3.1
3.1.1
3.2
3.2.1
3.3. ___ حاجز الجدار
3.4 .___ إشارة
3.4.1
3.5. ___ قائمة انتظار
3.6 .___ فئات المزامنة الأخرى
4. تطوير الصف للتزامن
4.1. ___ الفصول التي تستخدم في انتظار
4.1.1
4.1.2
4.2. ___ التكرارات غير المتزامنة
4.3. ___ مدراء السياق غير المتزامن
5. استثناءات من المهلات وبسبب إلغاء المهمة
5.1. ___ استثناءات
5.2. ___ استثناءات بسبب المهل وبسبب إلغاء المهام
5.2.1
5.2.2 .______ Coroutines مع مهلات
6. التفاعل مع الأجهزة
6.1. ___ مشاكل المزامنة
6.2. ___ أجهزة الاقتراع مع coroutines
6.3. __ باستخدام محرك الدفق
6.3.1
6.4 .___ تطوير برنامج التشغيل لجهاز الدفق
6.5. ___ مثال كامل: برنامج تشغيل aremote.py لجهاز الاستقبال للتحكم عن بعد بالأشعة تحت الحمراء.
6.6 .___ سائق جهاز استشعار درجة الحرارة والرطوبة HTU21D.
7. نصائح والخدع
7.1. ___ البرنامج يتجمد
7.2. __ uasyncio يحفظ الحالة
7.3. ___ جمع القمامة
7.4
7.5. ___ خطأ شائع. يمكن أن يكون من الصعب العثور عليها.
7.6. ___ البرمجة باستخدام مآخذ ( مآخذ )
7.6.1
7.7. __ ​​وسيطات منشئ حلقة الحدث
8. ملاحظات للمبتدئين
8.1. ___ المشكلة 1: حلقات الحدث
8.2. ___ المشكلة 2: طرق القفل
8.3. __ نهج uasyncio
8.4 .___ التخطيط في uasyncio
8.5 .___ لماذا الجدولة التعاونية ، وليس القائمة على سلسلة الرسائل ( _thread )؟
8.6. ___ التفاعل
8.7. _ الاقتراع

0. مقدمة

يفترض معظم هذا المستند بعض الإلمام بالبرمجة غير المتزامنة. للمبتدئين ، يمكن العثور على مقدمة في القسم 7.

تتضمن مكتبة uasyncio لـ MicroPython مجموعة فرعية من مكتبة Asyncio Python وهي مخصصة للاستخدام على المتحكمات الدقيقة. على هذا النحو ، فإنه يأخذ كمية صغيرة من ذاكرة الوصول العشوائي ويتم تكوينه للتبديل بسرعة السياقات مع تخصيص ذاكرة الوصول العشوائي صفر.

يصف هذا المستند استخدام uasyncio مع التركيز على إنشاء برامج تشغيل للأجهزة.

الهدف من ذلك هو تصميم برامج التشغيل بحيث يستمر التطبيق في العمل بينما ينتظر برنامج التشغيل استجابة من الجهاز. في الوقت نفسه ، يظل التطبيق حساسًا للأحداث الأخرى وتفاعل المستخدم.

مجال هام آخر لتطبيق التزامن هو برمجة الشبكات: على الإنترنت ، يمكنك العثور على معلومات كافية حول هذا الموضوع.

لاحظ أن MicroPython يعتمد على Python 3.4 مع الحد الأدنى من الإضافات Python 3.5 . باستثناء ما هو مفصل أدناه ، لا يتم دعم وظائف الإصدارات المتزامنة الأقدم من 3.4. يحدد هذا المستند الميزات المدعومة في هذه المجموعة الفرعية.

الغرض من هذا الدليل هو تقديم أسلوب برمجة متوافق مع CPython V3.5 والإصدارات الأحدث.

0.1 تثبيت uasyncio على جهاز فارغ (الأجهزة)

يوصى باستخدام البرامج الثابتة MicroPython V1.11 أو الأحدث. في العديد من الأنظمة الأساسية ، التثبيت غير مطلوب ، نظرًا لأن uasyncio مترجم بالفعل في التجميع. للتحقق ، فقط اكتب في REPL

import uasyncio 

تغطي الإرشادات التالية حالات عدم تثبيت الوحدات مسبقًا. قوائم الانتظار ووحدات التزامن اختيارية ، ولكنها مطلوبة لتشغيل الأمثلة الواردة هنا.

جهاز متصل بالإنترنت

على جهاز متصل بالإنترنت وتشغيل البرامج الثابتة V1.11 أو الأحدث ، يمكنك التثبيت باستخدام إصدار upup المدمج . التأكد من أن الجهاز متصل بشبكتك:

 import upip upip.install ( 'micropython-uasyncio' ) upip.install ( 'micropython-uasyncio.synchro' ) upip.install ( 'micropython-uasyncio.queues' ) 

رسائل الخطأ من upip ليست مفيدة للغاية. إذا حصلت على خطأ غير مفهوم ، تحقق من اتصال الإنترنت مرة أخرى.

أجهزة بدون اتصال بالإنترنت ( micropip )

إذا لم يكن لدى الجهاز اتصال بالإنترنت (على سبيل المثال ، Pyboard V1.x ) ، فإن أسهل طريقة هي بدء تثبيت micropip.py على الكمبيوتر إلى الدليل الذي تختاره ، ثم نسخ بنية الدليل الناتج إلى الجهاز الهدف. تعمل الأداة المساعدة micropip.py على Python 3.2 أو إصدار أحدث وتعمل على أنظمة Linux و Windows و OSX. يمكن الاطلاع على مزيد من المعلومات هنا .

مكالمة نموذجية:

 $ micropip.py install -p ~/rats micropython-uasyncio $ micropip.py install -p ~/rats micropython-uasyncio.synchro $ micropip.py install -p ~/rats micropython-uasyncio.queues 

جهاز بدون اتصال بالإنترنت (مصدر النسخ)

إذا كنت لا تستخدم micropip.py ، فيجب نسخ الملفات من المصدر. تصف الإرشادات التالية كيفية نسخ الحد الأدنى لعدد الملفات إلى الجهاز المستهدف ، وكذلك الحالة التي يحتاج فيها ضغط uasyncio إلى مجموعة مترجمة في شكل رمز ثانوي لتقليل المساحة المشغولة. للحصول على أحدث إصدار متوافق مع البرامج الثابتة الرسمية ، يجب نسخ الملفات من موقع micropython-lib الرسمي.

استنساخ المكتبة إلى الكمبيوتر باستخدام الأمر

 $ git clone https://github.com/micropython/micropython-lib.git 

على الجهاز الهدف ، أنشئ دليل uasyncio (اختياري في دليل lib) وانسخ الملفات التالية فيه:

• uasyncio / uasyncio / __ init__.py
• uasyncio.core / uasyncio / core.py
• uasyncio.synchro / uasyncio / synchro.py
• uasyncio.queues / uasyncio / queues.py


يمكن ضغط وحدات uasyncio هذه في رمز bytecode عن طريق وضع دليل uasyncio ومحتوياته في منفذ دليل الوحدات النمطية وإعادة ترجمة المحتويات.

1. التخطيط المشترك

يتم استخدام تقنية التنفيذ المشترك للعديد من المهام على نطاق واسع في الأنظمة المدمجة ، والتي توفر مقدار حمل أقل من جدولة الخيوط ( _thread ) ، وتجنب العديد من المزالق المرتبطة بسلسلة مؤشرات الترابط غير المتزامنة بالفعل.

1.1 وحدات

فيما يلي قائمة بالوحدات النمطية التي يمكن تشغيلها على الجهاز الهدف.

المكتبات

1. asyn.py يوفر تأمين ، حدث ، حاجز ، إشارة ، BoundedSemaphore ، شرط ، جمع بدائل التزامن. يوفر الدعم لإلغاء المهام من خلال فصول NamedTask و Cancellable .

2. aswitch.py يمثل الطبقات لمفاتيح التبديل والأزرار ، وكذلك كائن البرنامج مع إمكانية التأخير المتكرر. الأزرار هي تعميم للمفاتيح التي توفر حالة منطقية بدلاً من الحالة المادية ، وكذلك الأحداث التي تحدث بسبب الضغط المزدوج والطويل.

البرامج التجريبية

أولهما مفيد للغاية حيث أنهما يعطيان نتائج مرئية عند الوصول إلى أجهزة Pyboard .

1. وميض aledflash.py أربعة مؤشرات Pyboard بشكل غير متزامن لمدة 10 ثوانٍ. أبسط مظاهرة uasyncio . استيراده لتشغيل.
2. apoll.py برنامج تشغيل الجهاز الخاص بتسارع Pyboard . يوضح استخدام coroutines للاستعلام عن الجهاز. يعمل لمدة 20 ثانية. استيراده لتشغيل. يتطلب Pyboard V1.x.
3. astests.py برامج اختبار / تجريبي لوحدة aswitch .
4. asyn_demos.py عروض بسيطة لإلغاء المهام.
5. roundrobin.py مظاهرة التخطيط الدائري. أيضا معيار لتخطيط الأداء.
6. awaitable.py مظاهرة من فئة مع الانتظار. طريقة واحدة لتنفيذ برنامج تشغيل جهاز يستقصي واجهة.
7. chain.py نسخ من وثائق بيثون . مظاهرة سلسلة كوروتين.
8. aqtest.py عرض توضيحي لفئة قائمة الانتظار في مكتبة uasyncio .
9. aremote.py مثال برنامج تشغيل الجهاز لبروتوكول NEC IR.
10. auart.py مظاهرة تدفق المدخلات والمخرجات من خلال Pyboard UART .
11. auart_hd.py استخدام Pyboard UART للاتصال بجهاز باستخدام بروتوكول أحادي الاتجاه. مناسب للأجهزة ، على سبيل المثال ، باستخدام مجموعة أوامر AT modem.
12. iorw.py عرض توضيحي لجهاز قارئ / كاتب باستخدام تدفق I / O.

اختبار البرامج

1. asyntest.py اختبارات لفئات المزامنة في asyn.py.
2. اختبارات إلغاء الوظيفة.

فائدة

1. check_async_code.py تتم كتابة الأداة المساعدة في Python3 لاكتشاف أخطاء الترميز المحددة التي يصعب العثور عليها. انظر القسم 7.5.

سيطرة

يحتوي دليل المعايير على برامج نصية لفحص وتوصيف برنامج جدولة uasyncio .


2. مكتبة uasyncio

يعتمد مفهوم التزامن على تنظيم التخطيط للتنفيذ المشترك للعديد من المهام ، والتي تسمى في هذه الوثيقة coroutines .

2.1 هيكل البرنامج: حلقة الحدث

النظر في المثال التالي:

 import uasyncio as asyncio async def bar (): count = 0, while True : count + = 1 print ( count ) await asyncio.sleep ( 1 ) #  1 loop = asyncio.get_event_loop () loop.create_task ( bar ()) #     loop.run_forever () 

يستمر تنفيذ البرنامج حتى يتم استدعاء loop.run_forever . في هذه المرحلة ، يتم التحكم في التنفيذ بواسطة المجدول. لن يتم تنفيذ السطر بعد loop.run_forever أبدًا. يقوم برنامج الجدولة بتنفيذ الرمز الشريطي لأنه تم وضعه في قائمة الانتظار في برنامج جدولة loop.create_task . في هذا المثال التافه ، هناك شريط coroutine واحد فقط. إذا كان هناك آخرون ، فسوف ينفذهم المجدول خلال الفترات التي تم فيها إيقاف الشريط مؤقتًا.

تحتوي معظم التطبيقات المضمنة على حلقة حدث مستمر. يمكن أيضًا بدء حلقة حدث بطريقة تسمح بالإكمال باستخدام طريقة حلقة الحدث run_until_complete ؛ وهي تستخدم أساسا في الاختبار. يمكن العثور على أمثلة في وحدة astests.py .

مثيل حلقة الحدث هو كائن واحد تم إنشاؤه بواسطة الاستدعاء الأول إلى asyncio.get_event_loop () مع وسيطة عدد صحيح اختياري تشير إلى عدد coroutines في قوائم الانتظار اثنين - البداية والانتظار. عادة ، سيكون لكلتا الوسيطتين نفس القيمة ، مساوية لعدد على الأقل من coroutines المنفذة في وقت واحد في التطبيق. عادةً ما تكون القيمة الافتراضية 16 كافية. في حالة استخدام القيم غير الافتراضية ، راجع وسيطات مُنشئ حلقة الحدث (القسم 7.7.).

إذا كان coroutine يحتاج إلى استدعاء أسلوب حلقة الحدث (عادةً إنشاء_t_task ) ، فإن استدعاء asyncio.get_event_loop () (بدون وسيطات) سيعيده بشكل فعال.

2.2 Coroutines

يتم إنشاء coroutine على النحو التالي:

 async def foo ( delay_secs ): await asyncio.sleep ( delay_secs ) print ( 'Hello' ) 

يمكن ل coroutine أن يسمح بتشغيل coroutines أخرى باستخدام بيان الانتظار . يجب أن يحتوي coroutine على بيان انتظار واحد على الأقل. هذا يؤدي إلى تنفيذ coroutine قبل الانتهاء ، قبل تنفيذ التنفيذ إلى العبارة التالية. النظر في مثال:

 await asyncio.sleep ( delay_secs ) await asyncio.sleep ( 0 ) 

يؤدي السطر الأول إلى توقف الشفرة مؤقتًا لفترة تأخير ، في حين يستخدم كوريون آخرون هذا الوقت لتنفيذها. يؤدي التأخير 0 إلى تنفيذ جميع الخطوط المعلقة في ترتيب دوري حتى يتم تنفيذ السطر التالي. انظر مثال roundrobin.py .

2.2.1. قائمة انتظار للتخطيط coroutine

• EventLoop.create_task وسيطة: Coroutine لتشغيل. يصطف المجدول قائمة الانتظار للبدء في أسرع وقت ممكن. استدعاء create_task يعود على الفور. يتم تحديد coroutine في الوسيطة باستخدام بناء جملة استدعاء الدالة مع الوسائط الضرورية.
• EventLoop.run_until_complete وسيطة: Coroutine لتشغيل. يصطف المجدول قائمة الانتظار للبدء في أسرع وقت ممكن. يتم تحديد coroutine في الوسيطة باستخدام بناء جملة استدعاء الدالة مع الوسائط الضرورية. إرجاع المكالمة un_until_complete عند اكتمال coroutine: توفر هذه الطريقة طريقة لإنهاء برنامج الجدولة.
• في انتظار وسيطة: coroutine لتشغيل ، المحدد باستخدام بناء جملة استدعاء الدالة. يبدأ coroutine في أقرب وقت ممكن. يتم حظر coroutine المعلقة حتى يكتمل أحد coroutines المتوقع.

ما سبق متوافق مع CPython . وتناقش أساليب uasyncio إضافية في الملاحظات (القسم 2.2.3.).

2.2.2 بدء وظيفة رد الاتصال

يجب أن تكون عمليات الاسترجاعات وظائف Python مصممة للتنفيذ في فترة زمنية قصيرة. هذا يرجع إلى حقيقة أن coroutines لن تكون قادرة على العمل طوال مدة تنفيذ هذه الوظيفة.

استخدام أساليب فئة EventLoop التالية رد الاتصال:

1. call_soon - استدعاء في أقرب وقت ممكن. Args: رد الاتصال callback لتشغيل ، * args يمكن أن يتبع أي وسيطات موضعية بفاصلة.
2. call_later - استدعاء بعد تأخير في ثوان. Args: التأخير ، رد الاتصال ، * args
3. call_later_ms - مكالمة بعد تأخير في مللي ثانية. Args: التأخير ، رد الاتصال ، * args .

 loop = asyncio.get_event_loop () loop.call_soon ( foo , 5 ) #    'foo'      5. loop.call_later ( 2 , foo , 5 ) #   2 . loop.call_later_ms ( 50 , foo , 5 ) #   50 . loop.run_forever () 

2.2.3 ملاحظات

قد يحتوي coroutine على عبارة إرجاع مع قيم الإرجاع التعسفي. للحصول على هذه القيمة:

 result = await my_coro () 

يمكن أن يكون coroutine محدودًا بالطرق ويجب أن يحتوي على بيان انتظار واحد على الأقل.

2.3 التأخير

هناك خياران لتنظيم التأخير في coroutines. للتأخيرات الأطول وفي الحالات التي لا تحتاج فيها إلى أن تكون المدة دقيقة ، يمكنك استخدام:

 async def foo( delay_secs , delay_ms ): await asyncio.sleep ( delay_secs ) print ( 'Hello' ) await asyncio.sleep_ms ( delay_ms ) 

خلال مثل هذه التأخيرات ، سوف يقوم المجدول بتنفيذ برامج أخرى. يمكن أن يؤدي هذا إلى عدم اليقين في الوقت ، حيث لن يتم تشغيل coroutine الاستدعاء إلا عند تنفيذ الذي يتم تشغيله حاليًا. يعتمد مقدار التأخير على مطور التطبيق ، ولكن من المحتمل أن يكون حسب ترتيب عشرات أو مئات مللي ثانية ؛ تمت مناقشة ذلك بشكل أكبر في التفاعل مع أجهزة الأجهزة (القسم 6).

يمكن إجراء تأخير دقيق للغاية باستخدام وظائف utime - sleep_ms و sleep_us . هي الأنسب للتأخيرات القصيرة ، حيث أن برنامج الجدولة لن يكون قادرًا على تنفيذ برامج أخرى أثناء التأجيل.

3.Sinhronizatsiya

في كثير من الأحيان هناك حاجة لضمان التزامن بين coroutines. مثال شائع هو تجنب ما يسمى "شروط السباق" عندما تتطلب العديد من coroutines في وقت واحد الوصول إلى نفس المورد. يتم توفير مثال في astests.py وتناقش في الوثائق . خطر آخر هو "العناق الموت" عندما ينتظر كل coroutine الآخر لإكمال.

في التطبيقات البسيطة ، يمكن تحقيق التزامن باستخدام الأعلام العالمية أو المتغيرات ذات الصلة. أسلوب أكثر أناقة هو استخدام فئات التزامن. تقدم الوحدة النمطية asyn.py تطبيقات "متناهية الصغر" لفئات الأحداث ، والحاجز ، والإشارة ، وكونديتيوس ، المعدة للاستخدام مع المتزامن فقط. فهي ليست موجهة نحو مؤشر الترابط ويجب عدم استخدامها مع الوحدة النمطية _thread أو معالج المقاطعة ، ما لم ينص على خلاف ذلك. يتم تطبيق فئة القفل أيضًا ، وهو بديل للتطبيق الرسمي.

تنشأ مشكلة تزامن أخرى مع منتجي coroutine والمستهلكين coroutine. منتج coroutine يولد البيانات التي يستخدمها المستهلك coroutine. للقيام بذلك ، يوفر asyncio فئة قائمة الانتظار . يضع منتج coroutine البيانات في قائمة الانتظار ، بينما ينتظر المستهلك coroutine إكماله (مع عمليات أخرى مجدولة في الوقت المحدد). توفر فئة قائمة الانتظار ضمانات لإزالة العناصر بالترتيب الذي وردت به. بدلاً من ذلك ، يمكنك استخدام فئة Barrier إذا كان يجب على coroutine المنتج الانتظار حتى يصبح coroutine المستهلك جاهزًا للوصول إلى البيانات.

ويرد لمحة موجزة عن الطبقات أدناه. مزيد من التفاصيل في الوثائق الكاملة .

3.1 قفل

يضمن قفل الوصول الفريد إلى مورد مشترك. يقوم مثال التعليمة البرمجية التالي بإنشاء مثيل لقفل تأمين الفئة الذي يتم تمريره إلى كافة العملاء الذين يرغبون في الوصول إلى المورد المشترك. يحاول كل coroutine التقاط القفل ، مع إيقاف التنفيذ مؤقتًا حتى ينجح:

 import uasyncio as asyncio from uasyncio.synchro import Lock async def task(i, lock): while 1: await lock.acquire() print("Acquired lock in task", i) await asyncio.sleep(0.5) lock.release() async def killer(): await asyncio.sleep(10) loop = asyncio.get_event_loop() lock = Lock() # The global Lock instance loop.create_task(task(1, lock)) loop.create_task(task(2, lock)) loop.create_task(task(3, lock)) loop.run_until_complete(killer()) #  10s 

3.1.1.Lock والمهلة

في وقت كتابة هذا التقرير (5 كانون الثاني (يناير) 2018 ، لم يتم إكمال تطوير فئة تأمين uasycio Lock رسميًا. إذا كان للكوروتيني مهلة (القسم 5.2.2.) ، عند انتظار القفل عند التشغيل ، فإن المهلة ستكون غير فعالة. لن تتلقى TimeoutError حتى يتلقى تأمين. الأمر نفسه ينطبق على إلغاء المهمة.

تقدم الوحدة النمطية asyn.py فئة التأمين ، والتي تعمل في هذه الحالات. هذا التطبيق للفئة أقل كفاءة من الفصل الرسمي ، لكنه يدعم واجهات إضافية وفقًا لإصدار CPython ، بما في ذلك استخدام مدير السياق.

3.2 الحدث

الحدث يخلق فرصة واحدة أو عدة coroutines لإيقاف ، في حين يعطي آخر إشارة حول استمرارها. يصبح مثيل الحدث متاحًا لجميع coroutines الذين يستخدمونه:

 import asyn event = asyn.Event () 

ينتظر coroutine حدثًا عن طريق إعلان حدث انتظار ، وبعد ذلك يتوقف التنفيذ مؤقتًا إلى أن يعلن coroutines آخر event.set () . معلومات كاملة

قد تنشأ مشكلة إذا تم إصدار event.set () في إنشاء حلقات ؛ يجب أن ينتظر الرمز حتى تتمكن جميع الكائنات المعلقة من الوصول إلى الحدث قبل ضبطه مرة أخرى. في الحالة التي يتوقع فيها coro حدوث حدث ، يمكن تحقيق ذلك من خلال تلقي حدث coro يمسح الحدث:

 async def eventwait ( event ): await event event.clear() 

يتحقق coroutine الذي يطلق الحدث من أنه قد تمت خدمته:

 async def foo ( event ): while True : #   - while event.is_set (): await asyncio.sleep ( 1 ) # ,  coro   event.set () 

في حالة انتظار العديد من coros لمزامنة حدث واحد ، يمكن حل المشكلة باستخدام حدث التأكيد. يحتاج كل coro إلى حدث منفصل.

 async def eventwait (  , ack_event ): await event ack_event.set () 

يتم إعطاء مثال على ذلك في دالة event_test في asyntest.py . هذا مرهق في معظم الحالات - حتى مع وجود coro انتظار واحد - تقدم فئة Barrier ، الموضحة أدناه ، طريقة أكثر بساطة.
يمكن أن يوفر الحدث أيضًا وسيلة اتصال بين معالج المقاطعة و coro . يحافظ المعالج على الأجهزة ويقوم بتعيين الحدث ، والذي يتم التحقق منه بواسطة coro بالفعل في الوضع العادي.

3.2.1 قيم الحدث

يمكن أن تأخذ طريقة event.set () قيمة بيانات اختيارية من أي نوع. يمكن Coro ، في انتظار حدث ، الحصول عليه مع event.value () . لاحظ أنه سيتم تعيين event.clear () إلى بلا. الاستخدام النموذجي لهذا الإعداد coro للحدث هو إصدار event.set (utime.ticks_ms ()) . يمكن لأي coro تنتظر حدث تحديد التأخير الذي حدث ، على سبيل المثال ، للتعويض عن هذا.

3.3 الحاجز

هناك نوعان من الاستخدامات لفئة الحاجز .

أولاً ، يمكنه تعليق coroutine حتى يتم الانتهاء من واحد أو عدة coroutines أخرى.

ثانياً ، يسمح للعديد من coroutines بالالتقاء عند نقطة معينة. على سبيل المثال ، يمكن للمنتج والمستهلك إجراء المزامنة عند النقطة التي يكون فيها للمنتج بيانات ، والمستهلك جاهز لاستخدامها. في وقت التنفيذ ، قد يصدر الجدار رد اتصال إضافي قبل إزالة الحاجز وقد تستمر جميع الأحداث المعلقة.

يمكن أن يكون رد الاتصال وظيفة أو coroutine. في معظم التطبيقات ، سيتم استخدام الوظيفة على الأرجح: يمكن ضمان أن يتم تنفيذها قبل الانتهاء ، قبل إزالة الحاجز.

مثال على ذلك هو barrier_test وظيفة في asyntest.py . في مقتطف الشفرة لهذا البرنامج:

 import asyn def callback(text): print(text) barrier = asyn.Barrier(3, callback, ('Synch',)) async def report(): for i in range(5): print('{} '.format(i), end='') await barrier 

تقوم عدة مثيلات للتقرير بطباعة نتيجتها وتوقفها مؤقتًا حتى تكتمل مثيلات أخرى وتنتظر استمرار الحاجز . في هذه المرحلة ، يتم إجراء رد اتصال. عند الانتهاء ، يستأنف coroutine الأصلي.

3.4 إشارة

يحد الإشارة من عدد coroutines التي يمكنها الوصول إلى المورد. يمكن استخدامه للحد من عدد حالات coroutine معينة التي يمكن أن تعمل في وقت واحد. يتم ذلك باستخدام عداد وصول ، يتم تهيئته بواسطة المنشئ ويتم تقليله في كل مرة يتلقى فيها coroutine إشارة.

أسهل طريقة لاستخدامها في مدير السياق:

 import asyn sema = asyn.Semaphore(3) async def foo(sema): async with sema: #    

مثال على ذلك هو وظيفة semaphore_test في asyntest.py .

3.4.1 ( محدود ) إشارة

إنه يعمل بشكل مشابه لفئة Semaphore فيما عدا أنه إذا تسببت طريقة الإصدار في تجاوز عداد الوصول عن قيمته الأولية ، يتم تعيين ValueError .

3.5 قائمة الانتظار

تتم المحافظة على فئة قائمة الانتظار بواسطة uasycio الرسمي ويوضح برنامج عينة aqtest.py استخدامه. يتم إنشاء قائمة الانتظار كالتالي:

 from uasyncio.queues import Queue q = Queue () 

coroutine الشركة المصنعة نموذجي يمكن أن تعمل على النحو التالي:

 async def producer(q): while True: result = await slow_process() #       await q.put(result) #  ,        

و coroutine المستهلك يمكن أن تعمل على النحو التالي:

 async def consumer(q): while True: result = await(q.get()) # ,  q  print('Result was {}'.format(result)) 

توفر فئة قائمة الانتظار وظائف إضافية هامة عندما يكون حجم قوائم الانتظار محدودًا ويمكن استقصاء الحالة. يمكن التحكم في السلوك باستخدام قائمة انتظار فارغة (إذا كان الحجم محدودًا) ويمكن التحكم في السلوك باستخدام قائمة انتظار كاملة.وثائق حول هذا في التعليمات البرمجية.

3.6 فئات التزامن الأخرى توفر

مكتبة asyn.py تطبيقًا صغريًا لبعض ميزات CPython الأخرى . تسمح

فئة الحالة لـ coroutine بإخطار coroutines الآخرين بانتظار مورد مغلق. بعد تلقي الإشعار ، سيتمكنون من الوصول إلى المورد وإلغاء قفله بدوره. قد يحد coroutine الإعلام من عدد coroutines ليتم إعلامك.

فئة Gather تتيح لك تشغيل قائمة من coroutines. عند الانتهاء من هذا الأخير ، سيتم إرجاع قائمة النتائج. يستخدم هذا التطبيق "الصغير" بناء جملة مختلف. المهلات يمكن تطبيقها على أي من coroutines.

4 تطوير فصول للتزامن

في سياق تطوير برامج تشغيل الأجهزة ، الهدف هو التأكد من أنها تعمل دون حظر. يجب أن يضمن برنامج التشغيل coroutine أن يتم تنفيذ coroutines الأخرى أثناء انتظار برنامج التشغيل للجهاز لتنفيذ عمليات الأجهزة. على سبيل المثال ، يجب أن تسمح المهمة التي تنتظر وصول البيانات إلى UART ، أو قيام المستخدم بالضغط على زر ، بجدولة الأحداث الأخرى حتى وقوع الحدث.

4.1 الفصول التي تستخدم انتظار الانتظار

يمكن أن يقوم coroutine بإيقاف التنفيذ مؤقتًا أثناء انتظار كائن يمكن انتظاره . ضمن CPython ، يتم إنشاء فئة انتظار مخصصة عن طريق تطبيق الأسلوب __await__ الخاصالذي يعود المولد. يتم استخدام الفئة التي تنتظر الانتظار على النحو التالي:

 import uasyncio as asyncio class Foo(): def __await__(self): for n in range(5): print('__await__ called') yield from asyncio.sleep(1) #     return 42 __iter__ = __await__ # .   async def bar(): foo = Foo() # Foo - awaitable  print('waiting for foo') res = await foo #   print('done', res) loop = asyncio.get_event_loop() loop.run_until_complete(bar()) 

حاليا MicroPython لا يدعم __await__ ( قضية # 2678 ) والحل لاستخدامها __iter__ . توفر السلسلة __iter__ = __await__ إمكانية النقل بين CPython و MicroPython . أمثلة التعليمات البرمجية، راجع الطبقات الحدث، الحاجز، الإلغاء، الحالة في asyn.py .

4.1.1 الاستخدام في مديري السياق

يمكن استخدام الكائنات المتوقعة في مديري السياق المتزامن أو غير المتزامن ، وتوفير الطرق الخاصة اللازمة. بناء الجملة:

 with await awaitable as a: #  'as'   #    async with awaitable as a: #    (.) #  - 

لتحقيق ذلك ، يجب على المولد __await__ إرجاع نفسه . يتم تمرير هذا إلى أي متغير في جملة كما ، ويسمح أيضا أساليب خاصة للعمل. راجع asyn.Condition و asyntest.condition_test حيث يستخدم فئة الحالة في انتظار ويمكن استخدامه في مدير سياق متزامن.

4.1.2 coroutine ننتظر في

لغة بيثون يتطلب __await__ كانت وظيفة المولد. في MicroPython ، المولدات و coroutines متطابقة ، وبالتالي فإن الحل هو استخدام العائد من coro (args) .

الغرض من هذا الدليل هو تقديم رمز محمول لـ CPython 3.5 أو إصدار أحدث. في CPython ، المولدات و coroutines تختلف في المعنى. في CPython ، يحتوي coroutine على طريقة خاصة __await__ يسترجعها المولد. هذا محمول:

 up = False #   MicroPython? try: import uasyncio as asyncio up = True #    sys.implementation.name except ImportError: import asyncio async def times_two(n): # Coro   await asyncio.sleep(1) return 2 * n class Foo(): def __await__(self): res = 1 for n in range(5): print('__await__ called') if up: # MicroPython res = yield from times_two(res) else: # CPython res = yield from times_two(res).__await__() return res __iter__ = __await__ async def bar(): foo = Foo() # foo is awaitable print('waiting for foo') res = await foo #   print('done', res) loop = asyncio.get_event_loop() loop.run_until_complete(bar()) 

لاحظ أن __await__ ، العائد من asyncio.sleep (1) مسموح به بواسطة CPython . ما زلت لا أفهم كيف يتحقق هذا.

4.2 أدوات التكرار

غير المتزامنة توفر أدوات التكرار غير المتزامنة وسيلة لإرجاع تسلسل محدد أو غير محدود من القيم ويمكن استخدامها كوسيلة لاسترداد عناصر البيانات المتسلسلة عندما تأتي من جهاز للقراءة فقط. يستدعي مكرر غير متزامن رمز غير متزامن في طريقته التالية . يجب على الفصل تلبية المتطلبات التالية:

• يحتوي على طريقة __aiter__ المعرفة في def غير المتزامن وإرجاع مكرر غير متزامن.
• يحتوي على أسلوب __anext__ ، وهو بحد ذاته عبارة عن coroutine - أي ، يتم تعريفه عبر def async def ويتضمن عبارة انتظار واحدة على الأقل . لإيقاف التكرار ، يجب رفع استثناء StopAsyncIteration .

يتم استرداد القيم التسلسلية باستخدام المزامنة كما هو موضح أدناه:

 class AsyncIterable: def __init__(self): self.data = (1, 2, 3, 4, 5) self.index = 0 async def __aiter__(self): return self async def __anext__(self): data = await self.fetch_data() if data: return data else: raise StopAsyncIteration async def fetch_data(self): await asyncio.sleep(0.1) #     if self.index >= len(self.data): return None x = self.data[self.index] self.index += 1 return x async def run(): ai = AsyncIterable() async for x in ai: print(x) 

4.3 مدراء السياق غير المتزامنين

يمكن تصميم الفئات لدعم مدراء السياق غير المتزامنين الذين لديهم إجراءات للدخول والخروج وهي برامج مشتركة. مثال على ذلك هو تأمين الطبقة المذكورة أعلاه. يحتوي على __aenter__ coroutine ، وهو أمر مطلوب منطقيا للتشغيل غير المتزامن. لدعم البروتوكول غير المتزامن لمدير السياق ، يجب أن تكون طريقة __aexit__ الخاصة بها أيضًا عبارة عن coroutine ، ويتحقق ذلك عن طريق تضمين انتظار asyncio.sleep (0) . يمكن الوصول إلى هذه الفئات من داخل coroutine باستخدام بناء الجملة التالي:

 async def bar ( lock ): async with lock: print ( « bar » ) 

كما هو الحال مع مديري السياق العاديين ، فإن طريقة الخروج مضمونة ليتم استدعاؤها عندما يكمل مدير السياق عمله ، كالمعتاد ، ومن خلال استثناء. لتحقيق هذا الهدف ، يتم استخدام الأساليب الخاصة __aenter__ و __aexit__ التي يجب تعريفها على أنها coroutines في انتظار كائن coroutine آخر أو قيد انتظار . هذا المثال مأخوذ من فئة التأمين :

  async def __aenter__(self): await self.acquire() # a coro    async def return self async def __aexit__(self, *args): self.release() #   await asyncio.sleep_ms(0) 

إذا كان المتزامن مع يحتوي على جملة متغير ، فسيحصل المتغير على القيمة التي يتم إرجاعها بواسطة __aenter__ .

لضمان السلوك الصحيح ، يجب أن تكون البرامج الثابتة V1.9.10 أو أحدث.

5. استثناءات من المهلات وبسبب إلغاء المهام

ترتبط هذه المواضيع: يشتمل uasyncio على إلغاء المهام وتطبيق مهلة لمهمة ، مع استثناء استثناء للمهمة بطريقة خاصة.

5.1 استثناءات

في حالة حدوث استثناء في coroutine ، يجب معالجته إما في هذا coroutine أو في coroutine في انتظار اكتماله. هذا يضمن أن الاستثناء لا ينطبق على المجدول. في حالة حدوث استثناء ، سيتوقف المجدول عن العمل بتمرير الاستثناء إلى التعليمات البرمجية التي بدأها المجدول. لذلك ، لتجنب توقف برنامج الجدولة ، يجب على coroutine الذي تم إطلاقه باستخدام loop.create_task () التقاط أي استثناءات في الداخل.

استخدام رمي أو قريب لرمي استثناء في coroutine أمر غير معقول. هذا يدمر uasyncio ، مما يؤدي إلى بدء coroutine وربما الخروج عندما لا يزال في قائمة انتظار التنفيذ.

يوضح المثال أعلاه هذا الموقف. إذا سمح للعمل حتى النهاية ، يعمل كما هو متوقع.

 import uasyncio as asyncio async def foo(): await asyncio.sleep(3) print('About to throw exception.') 1/0 async def bar(): try: await foo() except ZeroDivisionError: print('foo  -   0') # ! raise #     . except KeyboardInterrupt: print('foo was interrupted by ctrl-c') #   ! raise async def shutdown(): print('Shutdown is running.') #     await asyncio.sleep(1) print('done') loop = asyncio.get_event_loop() try: loop.run_until_complete(bar()) except ZeroDivisionError: loop.run_until_complete(shutdown()) except KeyboardInterrupt: print('Keyboard interrupt at loop level.') loop.run_until_complete(shutdown()) 

ومع ذلك ، يؤدي إصدار مقاطعة لوحة المفاتيح إلى حدوث استثناء في حلقة الحدث. وذلك لأن تنفيذ uasyncio.sleep يتم تمريره إلى حلقة الحدث. لذلك ، يجب أن تلتقط التطبيقات التي تتطلب رمزًا واضحًا استجابة لمقاطعة لوحة المفاتيح استثناءًا على مستوى حلقة الحدث.

5.2 الإلغاء والمهلة

كما ذكر أعلاه ، تعمل هذه الوظائف عن طريق طرح استثناء للمهمة بطريقة خاصة باستخدام طريقة MicroPython coroutine pend_throw الخاصة . كيف يعمل يعتمد على الإصدار. في الإصدار الرسمي uasyncio v.2.0 ، لا تتم معالجة استثناء حتى المهمة المجدولة التالية. هذا يفرض تأخيرًا إذا كانت المهمة تتوقع النومالمدخلات والمخرجات. المهلات قد تتجاوز الفترة الاسمية. لا يمكن تحديد مهمة التراجع عن المهام الأخرى عند اكتمال التراجع.

يوجد حلاً حلاً حاليًا.

• الحل البديل : توفر مكتبة asyn وسيلة لانتظار المهام أو مجموعات المهام لإلغاء. انظر إلغاء وظيفة (القسم 5.2.1.).
• توفر مكتبة بول سوكولوفسكي uasyncio v2.4 ، ولكن هذا يتطلب البرامج الثابتة Pycopy .
• Fast_io مكتبة uasyncio يحل هذه المشكلة في بيثون (أقل بطريقة أنيقة) وتشغيل البرامج الثابتة الرسمية.

التسلسل الهرمي للاستثناءات المستخدمة هنا هو Exception-CancelledError-TimeoutError .

5.2.1 إلغاء وظيفة توفر

uasyncio وظيفة إلغاء (coro) . هذا يعمل عن طريق رمي استثناء لاستخدام pend_throw coroutine . كما أنه يعمل مع coroutines المتداخلة. الاستخدام كما يلي:

 async def foo(): while True: #  -  10 secs await asyncio.sleep(10) async def bar(loop): foo_instance = foo() #   coro loop.create_task(foo_instance) # code omitted asyncio.cancel(foo_instance) 

إذا تم تشغيل هذا المثال تحت uasyncio V2.0 ، عندما شريط ستصدر إلغاء أنها لا تصبح نافذة المفعول حتى المقرر المقبل فو والحالات فو يمكن أن يكون تأخير لمدة تصل إلى 10 ثانية. سيحدث مصدر تأخير آخر إذا كان foo ينتظر إدخال / إخراج. أينما حدث التأخير ، لن يتمكن الشريط من تحديد ما إذا كان قد تم إلغاء foo . يهم في بعض حالات الاستخدام.

عند استخدام مكتبات Paul Sokolovsky أو fast_io ، يكفي استخدام sleep (0):

 async def foo(): while True: #  -  10 secs await asyncio.sleep(10) async def bar(loop): foo_instance = foo() #   coro loop.create_task(foo_instance) #    asyncio.cancel(foo_instance) await asyncio.sleep(0) #    

سوف يعمل هذا أيضًا في الإصدار uasyncio v2.0 إذا لم يعد foo (وأي foo coroutine معلقًا ) للنوم ولم ينتظر I / O.

يحدث السلوك الذي قد يفاجئ الإهمال عندما يُتوقع إلغاء coroutine الذي يتم تشغيله بواسطة create_task وفي وضع الاستعداد. النظر في هذا المقتطف:

 async def foo(): while True: #  -  10 secs await asyncio.sleep(10) async def foo_runner(foo_instance): await foo_instance print('   ') async def bar(loop): foo_instance = foo() loop.create_task(foo_runner(foo_instance)) #    asyncio.cancel(foo_instance) 

عند إلغاء foo ، تتم إزالته من قائمة انتظار المجدول ؛ لأنه يفتقر إلى بيان الإرجاع ، لا يستأنف إجراء الاستدعاء foo_runner أبدًا. يوصى دائمًا باستبعاد الاستثناء الموجود في أقصى نطاق للوظيفة للتراجع:

 async def foo(): try: while True: await asyncio.sleep(10) await my_coro except asyncio.CancelledError: return 

في هذه الحالة ، لا يحتاج my_coro إلى الاستثناء ، حيث سيتم نشره على قناة الاتصال والتقاطه هناك.

المذكرة.لا تستخدم أساليب قريبة أو رمي coroutine تقنيات عند استخدام جدولة coroutine. هذا يقوض جدولة ، مما اضطر coroutine لتنفيذ التعليمات البرمجية ، حتى لو لم يكن من المقرر. قد يكون لهذا عواقب غير مرغوب فيها.

5.2.2 Coroutines مع مهلة

مهلة تنفيذها باستخدام uasyncio طرق .wait_for () و .wait_for_ms () . أنها تأخذ coroutine والكمون في ثوان أو مللي ثانية ، على التوالي ، كوسائط. في حالة انتهاء المهلة ، سيتم طرح TimeoutError في coroutine باستخدام pend_throw. يجب اكتشاف هذا الاستثناء من قبل المستخدم أو المتصل. يعد هذا ضروريًا للسبب الموضح أعلاه: إذا انتهت المهلة ، فسيتم إلغاؤها. ما لم يتم اكتشاف الخطأ وإعادته ، فإن الطريقة الوحيدة التي يستطيع المتصل من خلالها متابعة الاستثناء نفسه.

عندما تم اكتشاف الاستثناء بواسطة coroutine ، كان لدي إخفاقات غير واضحة إذا لم يتم الاستثناء في النطاق الخارجي ، كما هو موضح أدناه:

 import uasyncio as asyncio async def forever(): try: print('Starting') while True: await asyncio.sleep_ms(300) print('Got here') except asyncio.TimeoutError: print('Got timeout') # And return async def foo(): await asyncio.wait_for(forever(), 5) await asyncio.sleep(2) loop = asyncio.get_event_loop() loop.run_until_complete(foo()) 

بدلاً من ذلك ، يمكنك اعتراض وظيفة الاتصال:

 import uasyncio as asyncio async def forever(): print('Starting') while True: await asyncio.sleep_ms(300) print('Got here') async def foo(): try: await asyncio.wait_for(forever(), 5) except asyncio.TimeoutError: pass print('Timeout elapsed.') await asyncio.sleep(2) loop = asyncio.get_event_loop() loop.run_until_complete(foo()) 

ملاحظة ل Uasyncio v2.0 .

لا ينطبق هذا على مكتبات Paul Sokolovsky أو fast_io .

إذا بدأ coroutine في انتظار asyncio.sleep (t) ، مع تأخير طويل t ، فلن تتم إعادة تشغيل coroutine حتى تنتهي صلاحية t . في حالة انقضاء المهلة قبل انتهاء النوم ، سيحدث TimeoutError عند إعادة تحميل coroutine - أي عندما تنتهي ر . في الوقت الحقيقي ومن منظور المتصل ، سيتم تأخير استجابة TimeoutError الخاصة به.

إذا كان هذا مهمًا للتطبيق ، فقم بإنشاء تأخير طويل أثناء انتظار فترة قصيرة في الحلقة. coroutineasyn.sleep يدعم هذا.

6 التفاعل مع المعدات

إن أساس التفاعل بين أحداث uasyncio والأحداث غير المتزامنة الخارجية هو الاقتراع. قد تستخدم الأجهزة التي تتطلب استجابة سريعة مقاطعة. لكن التفاعل بين روتين المقاطعة (ISR) و coroutine المستخدم سوف يعتمد على استطلاعات الرأي. على سبيل المثال ، قد يقوم ISR باستدعاء Event أو تعيين علامة عمومية ، في حين يقوم coroutine في انتظار نتيجة ما باستقصاء كائن في كل مرة يتم فيها جدولة طلب ما.

يمكن إجراء الاستجواب بطريقتين ، صريحة أو ضمنية. يتم الأخير باستخدام دفق I / Oآلية، وهو نظام لتدفق الأجهزة مثل UART و مآخذ . في أبسط استطلاع صريح ، قد يتكون الكود التالي:

 async def poll_my_device(): global my_flag #   ISR while True: if my_flag: my_flag = False # service the device await asyncio.sleep(0) 

بدلاً من علامة عمومية ، يمكنك استخدام متغير مثيل لفئة الأحداث أو مثيل لفئة تستخدم في انتظار . نناقش مسح واضح أدناه.

يتكون الاستقصاء الضمني من تطوير برنامج تشغيل يعمل كجهاز إدخال / إخراج متدفق ، مثل مقبس UART أو تيار الإدخال / الإخراج ، الذي يستقصي الأجهزة باستخدام نظام select.poll Python : نظرًا لأن الاقتراع يتم إجراؤه في C ، فهو أسرع وأكثر كفاءة من استطلاع واضح. تمت مناقشة استخدام الدفق I / O في القسم 6.3.

نظرًا لفعاليته ، يوفر الاستقصاء الضمني ميزة لأسرع برامج تشغيل أجهزة الإدخال / الإخراج: يمكن إنشاء برامج تشغيل دفق للعديد من الأجهزة التي لا تعتبر عادةً أجهزة دفق. تمت مناقشة هذا بمزيد من التفصيل في القسم 6.4.

6.1 مشكلات المزامنة

كلا الاستطلاعات الصريحة والضمنية تستند حاليًا إلى التخطيط الدوري. افترض أن I / O يعمل في وقت واحد مع coroutines N مخصصة ، كل منها يعمل مع صفر تأخير. عندما يتم تقديم I / O ، سيتم بعد ذلك الاستطلاع بمجرد جدولة جميع عمليات المستخدم. يجب مراعاة التأخير المقدر عند التصميم. قد تتطلب قنوات I / O التخزين المؤقت ، مع توفير معدات ISR في الوقت الفعلي من المخازن المؤقتة و coroutines ، وملء أو تحرير المخازن المؤقتة في وقت أبطأ.

من الضروري أيضًا التفكير في إمكانية تجاوز ذلك: هذا هو الحال عندما يحدث شيء يتم استجوابه بواسطة coroutine أكثر من مرة قبل أن يتم التخطيط له بالفعل بواسطة coroutine.

قضية توقيت أخرى هي دقة الكمون. إذا كانت القضايا coroutine

 await asyncio.sleep_ms ( t ) #   

يضمن برنامج الجدولة تعليق التنفيذ لمدة لا تقل عن t ms. قد يكون التأخير الفعلي أكبر من t ، والذي يعتمد على تحميل النظام الحالي. إذا كان هناك شركات أخرى في هذا الوقت تنتظر الانتهاء من التأخير غير الصفر ، فسيتم جدولة السطر التالي على الفور للتنفيذ. ولكن إذا كانت شركات أخرى لا تزال تنتظر التنفيذ (إما لأنها تسببت في تأخير صفري أو بسبب انقضاء وقتهم أيضًا) ، فقد يكون من المقرر تنفيذها في وقت مبكر. هذا يقدم عدم اليقين التزامن في وظائف sleep () و sleep_ms () . يمكن حساب قيمة الحالة الأسوأ لهذا الفائض من خلال جمع قيم وقت التشغيل لجميع هذه coroutines لتحديد وقت إرسال الحالة الأسوأ إلى المجدول.

يوفر الإصدار fast_io من uasyncio في هذا السياق طريقة للتأكد من أن تدفق I / O سيتم استطلاعه في كل تكرار من المجدول. من المأمول أن يقبل uasyncio الرسمي التعديلات ذات الصلة في الوقت المناسب.

6.2 استجواب الأجهزة باستخدام coroutines

هذا هو الأسلوب البسيط الأكثر ملاءمة للأجهزة التي يمكن استجوابها بسرعة منخفضة نسبيًا. هذا يرجع في المقام الأول إلى حقيقة أن الاستقصاء بفاصل استقصاء قصير (أو صفر) يمكن أن يؤدي إلى حقيقة أن coroutine يستهلك وقتًا للمعالج أكثر مما هو مرغوب فيه للوقوع في الفاصل الزمني. يوضح

المثال apoll.py هذا الأسلوب عن طريق الاستعلام عن مقياس التسارع Pyboardمع فاصل من 100 مللي ثانية. ينفذ التصفية البسيطة لتجاهل الضوضاء ، ويطبع رسالة كل ثانيتين في حالة عدم حدوث حركة. يوفر

المثال aswitch.py برامج تشغيل لأجهزة التبديل وأجهزة الأزرار.

فيما يلي مثال لبرنامج التشغيل لجهاز قادر على القراءة والكتابة. لسهولة الاختبار ، يقوم Pyboard UART 4 ​​بمحاكاة الجهاز الشرطي. ينفذ السائق RecordOrientedUart Class، حيث يتم توفير البيانات في سجلات متغيرة الطول تتكون من مثيلات البايت. يضيف الكائن محددًا قبل إرسال البيانات الواردة وتخزينها مؤقتًا حتى يتم تلقي محدد إضافي. هذا مجرد عرض تجريبي وغير فعال لاستخدام UART مقارنة بتدفق الإدخال / الإخراج.

لإظهار النقل غير المتزامن ، نفترض أن الجهاز الذي تمت محاكاته لديه وسيلة للتحقق من اكتمال عملية النقل وأن التطبيق يتطلب منا الانتظار. لا يوجد أي من الافتراضات صحيح في هذا المثال ، لكن التعليمة البرمجية تمزيقها بانتظار asyncio.sleep (0.1) .

للبدء ، لا تنس توصيل مخرجات Pyboard X1 و X2 (UART Txd و Rxd)

 import uasyncio as asyncio from pyb import UART class RecordOrientedUart(): DELIMITER = b'\0' def __init__(self): self.uart = UART(4, 9600) self.data = b'' def __iter__(self): # Not __await__ issue #2678 data = b'' while not data.endswith(self.DELIMITER): yield from asyncio.sleep(0) # ,  : while not self.uart.any(): yield from asyncio.sleep(0) # timing may mean this is never called data = b''.join((data, self.uart.read(self.uart.any()))) self.data = data async def send_record(self, data): data = b''.join((data, self.DELIMITER)) self.uart.write(data) await self._send_complete() #          #        await asyncio.sleep(0) async def _send_complete(self): await asyncio.sleep(0.1) def read_record(self): # Synchronous: await the object before calling return self.data[0:-1] # Discard delimiter async def run(): foo = RecordOrientedUart() rx_data = b'' await foo.send_record(b'A line of text.') for _ in range(20): await foo #  coros       foo rx_data = foo.read_record() print('Got: {}'.format(rx_data)) await foo.send_record(rx_data) rx_data = b'' loop = asyncio.get_event_loop() loop.run_until_complete(run()) 

6.3 استخدام آلية التدفق ( الدفق )

يوضح المثال المدخلات والمخرجات المتزامنة على UART واحد من المعالج الصغري Pyboard .

للبدء ، قم بتوصيل مخرجات Pyboard X1 و X2 (UART Txd و Rxd)

 import uasyncio as asyncio from pyb import UART uart = UART(4, 9600) async def sender(): swriter = asyncio.StreamWriter(uart, {}) while True: await swriter.awrite('Hello uart\n') await asyncio.sleep(2) async def receiver(): sreader = asyncio.StreamReader(uart) while True: res = await sreader.readline() print('Received', res) loop = asyncio.get_event_loop() loop.create_task(sender()) loop.create_task(receiver()) loop.run_forever() 

يمكن العثور على رمز الدعم في __init__.py في مكتبة uasyncio . آلية تعمل بشكل جيد مثل برنامج تشغيل جهاز (مكتوب في C ) تنفذ الطرق التالية: IOCTL، قراءة، يقوم readline و الكتابة . القسم 6.4: كتابة برنامج تشغيل جهاز الجري يكشف تفاصيل حول كيفية كتابة برامج التشغيل هذه في بيثون .

يمكن UART تلقي البيانات في أي وقت. تقوم آلية I / O المتدفقة بالتحقق من الحروف الواردة المعلقة كلما تمكن برنامج الجدولة من التحكم. عندما يتم تشغيل coroutine ، الروتين المقاطعة المقاطعة الأحرف الواردة؛ سيتم حذفها عندما يفسح coroutine الطريق إلى برنامج الجدولة. لذلك ، يجب تصميم تطبيقات UART بحيث يقلل coroutines الوقت بين عمليات النقل إلى برنامج الجدولة من أجل تجنب تجاوزات المخزن المؤقت وفقدان البيانات. يمكن تحسين هذا باستخدام مخزن مؤقت أكبر قراءة UART أو معدل بيانات أقل. بدلاً من ذلك ، سيوفر التحكم في تدفق الأجهزة حلاً إذا كان مصدر البيانات يدعمه.

6.3.1 مثال على UART سائق

برنامج auart_hd.pyيوضح طريقة اتصال بجهاز أحادي الاتجاه ، مثل جهاز يستجيب لمجموعة أوامر AT modem. يشير مصطلح "نصف مزدوج" إلى أن الجهاز لا يرسل أبدًا بيانات غير مطلوبة: يتم تنفيذ عمليات النقل دائمًا استجابةً لأمر تم استلامه من السيد.

يتم محاكاة الجهاز عن طريق إجراء اختبار على Pyboard مع اتصالين سلكيين .

يستجيب الجهاز (المبسط جدًا) المضاهاة لأي أمر عن طريق إرسال أربعة سطور من البيانات مع توقف مؤقت بين كل منهما لمحاكاة المعالجة البطيئة.

يرسل المعالج أمرًا ، لكنه لا يعرف مقدمًا عدد صفوف البيانات التي سيتم إرجاعها. إنه يبدأ مؤقت إعادة التشغيل الذي يتم إعادة تشغيله في كل مرة يتم فيها تلقي سطر. عند انتهاء الموقت ، من المفترض أن يكون الجهاز قد اكتمل الإرسال ويتم إرجاع قائمة الخطوط المستلمة.

يتم أيضًا إظهار حالة فشل الجهاز ، والتي يتم تحقيقها عن طريق تخطي الإرسال قبل انتظار الرد. بعد انتهاء المهلة ، يتم إرجاع قائمة فارغة. انظر التعليقات رمز لمزيد من التفاصيل.

6.4 تطوير تدفق (السائقين ستريم ) وحدة

تيار الإدخال / الإخراج آلية ( تيار I / O ) للسيطرة على عملية تدفق I / O الأجهزة مثل UART ومقابس ( المقبس). يمكن استخدام هذه الآلية بواسطة برامج تشغيل أي جهاز تم استطلاع آرائه بانتظام عن طريق تفويض المجدول الذي يستخدم التحديد ، والاستقصاء عن استعداد أي أجهزة في قائمة الانتظار. هذا أكثر فاعلية من إجراء العديد من عمليات coroutine ، حيث تقوم كل منها باستطلاعات الجهاز ، ويرجع ذلك جزئيًا إلى أن تحديد هو مكتوب في C ، وأيضًا بسبب تأجيل coroutine الذي يقوم بإجراء الاستقصاء حتى يعود الكائن الذي تم استطلاعه إلى حالة جاهزة.

يفضل أن يدعم برنامج تشغيل الجهاز القادر على خدمة آلية الإدخال / الإخراج المتدفق الطرق StreamReader، StreamWriter. يجب أن يوفر الجهاز القابل للقراءة إحدى الطرق التالية على الأقل. يرجى ملاحظة أن هذه هي طرق متزامنة. تضمن طريقة ioctl (انظر أدناه) أن يتم استدعاؤها فقط عند توفر البيانات. يجب إرجاع الطرق بأسرع وقت ممكن ، باستخدام أكبر قدر ممكن من البيانات المتاحة.

readline () إرجاع أكبر عدد ممكن من الأحرف ، حتى أي حرف سطر جديد. مطلوب إذا كان استخدام StreamReader.readline ()

read (n) إرجاع أكبر عدد ممكن من الأحرف ، لكن ليس أكثر من n . مطلوب في حالة استخدام StreamReader.read () أو StreamReader.readexactly ()

يجب أن يوفر برنامج التشغيل الذي تم إنشاؤه الطريقة المتزامنة التالية مع الإرجاع الفوري:

الكتابة باستخدام الوسائط buf ، off ، sz .

حيث:

buf هو المخزن المؤقت للكتابة.
قبالة - الإزاحة في المخزن المؤقت من الحرف الأول إلى الكتابة.
سز - العدد المطلوب من الشخصيات لكتابة.
قيمة الإرجاع هي عدد الأحرف المكتوبة بالفعل (ربما 1 إذا كان الجهاز بطيئًا). تضمن
طريقة ioctl أنه لن يتم استدعاؤه إلا عندما يكون الجهاز جاهزًا لاستقبال البيانات.

يجب أن توفر جميع الأجهزة طريقة ioctl تقوم باستقصاء المعدات لتحديد حالة توفرها. مثال نموذجي لبرنامج تشغيل القراءة / الكتابة:

 import io MP_STREAM_POLL_RD = const(1) MP_STREAM_POLL_WR = const(4) MP_STREAM_POLL = const(3) MP_STREAM_ERROR = const(-1) class MyIO(io.IOBase): #    def ioctl(self, req, arg): # see ports/stm32/uart.c ret = MP_STREAM_ERROR if req == MP_STREAM_POLL: ret = 0 if arg & MP_STREAM_POLL_RD: if hardware_has_at_least_one_char_to_read: ret |= MP_STREAM_POLL_RD if arg & MP_STREAM_POLL_WR: if hardware_can_accept_at_least_one_write_character: ret |= MP_STREAM_POLL_WR return ret 

فيما يلي وصف لتأخير انتظار MillisecTimer Class :

 import uasyncio as asyncio import utime import io MP_STREAM_POLL_RD = const(1) MP_STREAM_POLL = const(3) MP_STREAM_ERROR = const(-1) class MillisecTimer(io.IOBase): def __init__(self): self.end = 0 self.sreader = asyncio.StreamReader(self) def __iter__(self): await self.sreader.readline() def __call__(self, ms): self.end = utime.ticks_add(utime.ticks_ms(), ms) return self def readline(self): return b'\n' def ioctl(self, req, arg): ret = MP_STREAM_ERROR if req == MP_STREAM_POLL: ret = 0 if arg & MP_STREAM_POLL_RD: if utime.ticks_diff(utime.ticks_ms(), self.end) >= 0: ret |= MP_STREAM_POLL_RD return ret 

والتي يمكن استخدامها على النحو التالي:

 async def timer_test ( n ): timer = ms_timer.MillisecTimer () await timer ( 30 ) #  30  

بالمقارنة مع uasyncio الرسمية ، فإن هذا التطبيق لا يقدم أي مزايا مقارنة بانتظار asyncio.sleep_ms () . يوفر استخدام fast_io تأخيرات أكثر دقة بكثير في نمط الاستخدام العادي ، عندما يتوقع coroutines تأخير صفر.

يمكنك استخدام جدولة الإدخال / الإخراج لربط حدث مع رد اتصال. هذا أكثر كفاءة من دورة الاقتراع ، لأن الاقتراع غير مجد حتى عودة ioctl جاهزة. بعد ذلك ، يتم تنفيذ رد اتصال عندما تتغير حالة رد الاتصال.

 import uasyncio as asyncio import io MP_STREAM_POLL_RD = const(1) MP_STREAM_POLL = const(3) MP_STREAM_ERROR = const(-1) class PinCall(io.IOBase): def __init__(self, pin, *, cb_rise=None, cbr_args=(), cb_fall=None, cbf_args=()): self.pin = pin self.cb_rise = cb_rise self.cbr_args = cbr_args self.cb_fall = cb_fall self.cbf_args = cbf_args self.pinval = pin.value() self.sreader = asyncio.StreamReader(self) loop = asyncio.get_event_loop() loop.create_task(self.run()) async def run(self): while True: await self.sreader.read(1) def read(self, _): v = self.pinval if v and self.cb_rise is not None: self.cb_rise(*self.cbr_args) return b'\n' if not v and self.cb_fall is not None: self.cb_fall(*self.cbf_args) return b'\n' def ioctl(self, req, arg): ret = MP_STREAM_ERROR if req == MP_STREAM_POLL: ret = 0 if arg & MP_STREAM_POLL_RD: v = self.pin.value() if v != self.pinval: self.pinval = v ret = MP_STREAM_POLL_RD return ret 

ومرة أخرى - على uasyncio الرسمية ، قد يكون التأخير عالياً. اعتمادًا على تصميم التطبيق ، قد يكون إصدار fast_io أكثر كفاءة.

برنامج مظاهرة iorw.py يظهر مثال كاملة. يرجى ملاحظة أنه اعتبارا من كتابة هذه السطور في المسؤول uasyncio هناك خطأ، لأن من الذي لا يعمل . هناك حلان. يتمثل الحل البديل في كتابة برنامجين منفصلين ، أحدهما للقراءة فقط والآخر للكتابة فقط. والثاني هو استخدام fast_io ، الذي يحل هذه المشكلة.

في uasyncio الرسمية ، يتم تخطيط الإدخال / الإخراج نادراً ما .

6.5 المثال الكامل: aremote.py

تم تصميم برنامج التشغيل لاستقبال / فك تشفير الإشارات من جهاز التحكم عن بعد بالأشعة تحت الحمراء. سائق aremote.py نفسه . الملاحظات التالية مهمة فيما يتعلق باستخدام التزامن .

تسجل المقاطعة على جهة الاتصال وقت تغيير الحالة (بالميكروثانية) وتعيين الحدث ، وتخطي الوقت الذي حدث فيه تغيير الحالة الأولى. ينتظر coroutine لحدوث ما ، ويبلغ مدة حزمة البيانات ، ثم فك تشفير البيانات المخزنة قبل استدعاء رد الاتصال المحدد من قبل المستخدم.يتيح

مرور الوقت لمثيل الحدث أن يقوم coroutine بالتعويض عن أيتأخير غير متزامن عند تعيين فترة التأخير.

6.6 HTU21D استشعار البيئة

سائق يوفر رقاقة HTU21D القياس الدقيق لدرجة الحرارة والرطوبة.

تحتاج الشريحة إلى حوالي 120 مللي ثانية لتلقي كلاً من عناصر البيانات. يعمل برنامج التشغيل بشكل غير متزامن ، حيث يبدأ في استلام واستخدام انتظار asyncio.sleep (t) قبل قراءة البيانات ، ويقوم بتحديث متغيرات درجة الحرارة والرطوبة ، والتي يمكن الوصول إليها في أي وقت ، مما يتيح بدء تشغيل coroutines أخرى أثناء تشغيل برنامج تشغيل الرقاقة.

7. النصائح والخدع

7.1 البرنامج يتجمد

يحدث التجميد عادةً لأن المهمة محظورة بدون امتياز: سيؤدي ذلك إلى تجميد النظام بأكمله. عند التطوير ، من المفيد أن يكون لديك coroutine يعمل بشكل دوري على إضاءة LED المدمجة. هذا يوفر التأكيد على أن المجدول لا يزال قيد التشغيل.

7.2 يحفظ uasyncio الحالة

عند بدء تشغيل البرامج باستخدام uasyncio في REPL ، قم بإجراء إعادة تعيين ناعمة (ctrl-D) بين البدايات. نظرًا لحقيقة أن uasyncio يحافظ على الحالة بين البدايات ، فقد يحدث سلوك غير متوقع في البداية التالية.

7.3 جمع القمامة

يمكنك تنفيذ coroutine عن طريق تحديد الاستيراد gc أولاً :

 gc.collect () gc.treshold ( gc.mem_free () // 4 + gc.mem_alloc ()) 

تمت مناقشة الغرض من هذا هنا في قسم الكومة.

7.4 الاختبار من

المستحسن التأكد من أن برنامج تشغيل الجهاز يحتفظ بالتحكم عند الضرورة ، والذي يمكن القيام به من خلال تشغيل نسخة واحدة أو أكثر من النسخ التوضيحية الوهمية التي تبدأ دورة طباعة الرسائل والتأكد من تشغيلها أثناء فترات وجود برنامج التشغيل في وضع الاستعداد:

 async def rr(n): while True: print('Roundrobin ', n) await asyncio.sleep(0) 

كمثال على نوع الخطر الذي قد ينشأ ، في المثال أعلاه ، تمت كتابة طريقة RecordOrientedUart __await__ في الأصل على النحو التالي:

 def __await__(self): data = b'' while not data.endswith(self.DELIMITER): while not self.uart.any(): yield from asyncio.sleep(0) data = b''.join((data, self.uart.read(self.uart.any()))) self.data = data 

نتيجة لذلك ، يتم تنفيذ التنفيذ حتى يتم استلام السجل بالكامل ، وكذلك حقيقة أن uart.any () يُرجع دائمًا عددًا غير صفري من الأحرف المستلمة. بحلول وقت المكالمة ، ربما تكون جميع الأحرف قد تم استلامها بالفعل. يمكن حل هذا الموقف باستخدام حلقة خارجية:

 def __await__(self): data = b'' while not data.endswith(self.DELIMITER): yield from asyncio.sleep(0) # ,  : while not self.uart.any(): yield from asyncio.sleep(0) #        data = b''.join((data, self.uart.read(self.uart.any()))) self.data = data 

قد يكون من الجدير بالذكر أن هذا الخطأ لم يكن واضحًا إذا كانت البيانات قد تم إرسالها إلى UART بسرعة أقل ، بدلاً من استخدام اختبار الملاحظات. مرحبا بكم في أفراح البرمجة في الوقت الحقيقي.

7.5 خطأ شائع

إذا تم تعريف دالة أو طريقة بواسطة async def ثم تم استدعاؤها فيما بعد كما لو كانت مكالمة عادية (متزامنة) ، لا يعرض MicroPython رسالة خطأ. هذا حسب التصميم. عادة ما يؤدي هذا إلى حقيقة أن البرنامج بصمت لا يعمل بشكل صحيح:

 async def foo(): # code loop.create_task(foo) #  1 1: foo     foo() #  2: . 

لدي اقتراح يقترح إصلاح الموقف في الخيار 1 باستخدام fast_io . تحاول

وحدة check_async_code.py اكتشاف حالات الاستخدام المشكوك فيها للكوروتينات. هو مكتوب في Python3 ومصمم للعمل على جهاز كمبيوتر. يستخدم في نصوص مكتوبة وفقًا للإرشادات الموضحة في هذا الدليل مع coroutines المصرح باستخدام async def . الوحدة النمطية تأخذ وسيطة واحدة ، المسار إلى ملف MicroPython المصدر (أو - help ).

يرجى ملاحظة أنه غير مهذب إلى حد ما ويُقصد استخدامه في ملف صحيح بناءً ، والذي لا يبدأ افتراضيًا. استخدم أداة ، مثل pylint ، للتحقق من بناء الجملة العام ( pylint ليس لديها هذا الخطأ حاليًا).

السيناريو ينتج ايجابيات كاذبة. وفقًا للخطة ، فإن coroutines هي كائنات من المستوى الأول ؛ يمكن نقلها إلى وظائف وتخزينها في هياكل البيانات. اعتمادًا على منطق البرنامج ، يمكنك حفظ الوظيفة أو نتيجة تنفيذها. لا يمكن للبرنامج النصي تحديد القصد. يهدف إلى تجاهل الحالات التي تبدو صحيحة عند تحديد الحالات الأخرى التي يجب مراعاتها. لنفترض فو حيث يتم الإعلان عن coroutine غير متزامن def :

 loop.run_until_complete(foo()) #   bar(foo) #     ,      bar(foo()) z = (foo,) z = (foo(),) foo() #  :   . 

أجدها مفيدة كما هي ، لكن التحسينات مرحب بها دائمًا.

7.6 البرمجة مع مآخذ ( مآخذ )

هناك نهجين أساسيين لبرمجة مآخذ uasyncio . افتراضيًا ، يتم حظر المآخذ حتى تكتمل عملية القراءة أو الكتابة المحددة. يدعم Uasyncio قفل المقبس باستخدام select.poll لمنع المجدول من حظرها. في معظم الحالات ، تكون هذه الآلية أسهل في الاستخدام. يمكن العثور على مثال للعميل ورمز الخادم في دليل client_server . Userver استخدامات تطبيق select.poll الاستعلام صراحة مأخذ الخادم.

تستخدم مآخذ العميل بشكل ضمني بمعنى أن محرك تدفق uasyncio يستخدمه مباشرةً.

يرجى ملاحظة أن socket.getaddrinfo محظور حاليا. سيكون الحد الأدنى للوقت في نموذج التعليمة البرمجية ، ولكن إذا كان البحث عن DNS مطلوبًا ، فقد تكون فترة الحظر كبيرة.

الطريقة الثانية لبرمجة المقبس هي استخدام مآخذ التوصيل غير المحجوبة. يضيف هذا التعقيد ، لكنه ضروري في بعض التطبيقات ، خاصةً إذا كان الاتصال عبر WiFi (انظر أدناه).

في وقت كتابة هذا التقرير (مارس 2019) ، كان دعم TLS للمآخذ غير المقيدة قيد التطوير. وضعها الدقيق غير معروف (بالنسبة لي).

يتطلب استخدام مآخذ التوصيل غير المقيدة بعض الاهتمام بالتفاصيل. في حالة حدوث قراءات غير مانعة للحظر بسبب زمن الوصول للخادم ، فلا يوجد ما يضمن إعادة جميع (أو أي) البيانات المطلوبة. وبالمثل ، قد لا تنتقل الإدخالات.

لذلك ، يجب أن تقوم طرق القراءة والكتابة غير المتزامنة بإجراء عملية حظر غير متكررة حتى تتم قراءة البيانات المطلوبة أو كتابتها. في الممارسة العملية ، قد تكون هناك حاجة إلى مهلة للتعامل مع انقطاع الخادم.
المضاعفات الأخرى تتمثل في أن منفذ ESP32 لديه مشاكل تتطلب عمليات تفكك سيئة للغاية للتشغيل الخالي من الأخطاء. أنا لم تختبر ما إذا كان هذا لا يزال هو الحال.
وحدة Sock_nonblock.pyيوضح الطرق المطلوبة. هذا ليس عرضًا عمليًا ، ومن المحتمل أن تعتمد القرارات على التطبيق.

7.6.1 مشاكل مع شبكة WiFi ليست

آلية تدفق uasyncio هي الخيار الأفضل عند اكتشاف انقطاع شبكات WiFi. لقد وجدت أنه من الضروري استخدام مآخذ التوصيل غير المحظورة لتوفير عملية آمنة من الفشل وإعادة الاتصال بالعميل في حالة حدوث أعطال. يصف

هذا المستند المشكلات التي واجهتها في تطبيقات WiFi التي تبقي المقابس مفتوحة لفترات طويلة وتحدد الحل.

Pltcmيوفر عميل MQTT متزامن قوي يوفر تكامل الرسائل أثناء فشل WiFi. يوصف رابط تسلسلي مزدوج الاتجاه غير متزامن بسيط بين عميل لاسلكي وخادم سلكي مع تسليم الرسالة مضمونة.

7.7 وسيطات منشئ حلقة الحدث

قد يحدث خطأ صغير إذا كنت بحاجة إلى إنشاء حلقة حدث مع القيم التي تختلف عن القيم الافتراضية. يجب الإعلان عن هذه الحلقة قبل تشغيل أي كود آخر باستخدام التزامن لأن هذه القيم قد تكون مطلوبة في هذا الرمز. وإلا ، سيتم تهيئة الكود باستخدام القيم الافتراضية:

 import uasyncio as asyncio import some_module bar = some_module.Bar() #   get_event_loop() #     loop = asyncio.get_event_loop(runq_len=40, waitq_len=40) 

نظرًا لأن استيراد وحدة نمطية يمكنه تنفيذ التعليمات البرمجية ، فإن الطريقة الأكثر أمانًا هي إنشاء حلقة حدث على الفور بعد استيراد uasyncio .

 import uasyncio as asyncio loop = asyncio.get_event_loop(runq_len=40, waitq_len=40) import some_module bar = some_module.Bar() # get_event_loop()    

عند كتابة الوحدات النمطية للاستخدام بواسطة البرامج الأخرى ، أفضل تجنب تشغيل شفرة uasyncio عند الاستيراد. اكتب وظائف وطرق انتظار حلقة حدث كوسيطة. ثم تأكد من أن تطبيقات المستوى الأعلى هي فقط التي تتصل بـ get_event_loop :

 import uasyncio as asyncio import my_module #      loop = asyncio.get_event_loop(runq_len=40, waitq_len=40) bar = my_module.Bar(loop) 

تمت مناقشة هذه المشكلة هنا .

8 ملاحظات للمبتدئين

هذه الملاحظات مخصصة للمبتدئين في التعليمات البرمجية غير المتزامنة وتبدأ بوصفًا للمشاكل التي يحاول المخططون حلها ، بالإضافة إلى تقديم نظرة عامة على نهج uasyncio للحلول.

يناقش القسم 8.5 المزايا النسبية لوحدات uasyncio و _ thread ، وكذلك لماذا قد تفضل استخدام coroutines uasyncio بجدولة استباقية (_thread).

8.1 المشكلة 1: حلقات الحدث

يعمل تطبيق البرامج الثابتة النموذجي بشكل مستمر وفي الوقت نفسه يجب أن يستجيب للأحداث الخارجية ، والتي قد تشمل تغييرًا في الجهد على ADC أو ظهور مقاطعة للجهاز أو رمزًا تم استلامه في UART أو البيانات المتوفرة على المقبس. تحدث هذه الأحداث بشكل غير متزامن ، ويجب أن يكون الرمز قادراً على الاستجابة بغض النظر عن الترتيب الذي تحدث به. بالإضافة إلى ذلك ، قد تكون هناك حاجة إلى مهام تعتمد على الوقت ، مثل المصابيح الوامضة.

الطريقة الواضحة للقيام بذلك هي مع حلقة حدث uasycio . هذا المثال ليس رمزًا عمليًا ، ولكنه يخدم في توضيح الشكل العام لحلقة الحدث.

 def event_loop(): led_1_time = 0 led_1_period = 20 led_2_time = 0 led_2_period = 30 switch_state = switch.state() #    while True: time_now = utime.time() if time_now >= led_1_time: #  LED #1 led1.toggle() led_1_time = time_now + led_1_period if time_now >= led_2_time: #  LED #2 led2.toggle() led_2_time = time_now + led_2_period #    LEDs if switch.value() != switch_state: switch_state = switch.value() #  - if uart.any(): #    UART 

تعمل مثل هذه الحلقة على أمثلة بسيطة ، ولكن مع زيادة عدد الأحداث ، يصبح الرمز بسرعة مرهقًا. كما أنها تنتهك مبادئ البرمجة الموجهة للكائنات عن طريق الجمع بين معظم منطق البرنامج في مكان واحد ، بدلاً من ربط الكود بكائن متحكم فيه. نريد تطوير فئة لمصباح LED وميض يمكن إدراجه في وحدة نمطية واستيراده. قد تبدو طريقة OOP في وميض LED كما يلي:

 import pyb class LED_flashable(): def __init__(self, led_no): self.led = pyb.LED(led_no) def flash(self, period): while True: self.led.toggle() # -     period, #          

يسمح لك برنامج الجدولة في uasyncio بإنشاء هذه الفئات.

8.2 المشكلة 2: حظر الطرق

افترض أنك بحاجة إلى قراءة عدد معين من وحدات البايت من مأخذ التوصيل. إذا قمت بالاتصال بـ socket.read (n) بمقبس الحجب افتراضيًا ، فسيتم حظره (أي لن يكون قادرًا على الإنهاء) حتى يتم تلقي n بايت. خلال هذه الفترة ، لن يستجيب التطبيق لأحداث أخرى.

باستخدام مقبس uasyncio غير المحجوب ، يمكنك كتابة طريقة قراءة غير متزامنة. سيتم حظر المهمة التي تتطلب البيانات (بالضرورة) إلى أن يتم استلامها ، ولكن سيتم تنفيذ مهام أخرى خلال هذه الفترة ، مما سيسمح للتطبيق بالاستجابة.

8.3. طرق Uasyncio

تحتوي الفئة التالية على مؤشر LED يمكن تشغيله وإيقاف تشغيله ، كما يمكنك أيضًا وميض أي سرعة. يستخدم مثيل LED_async طريقة التشغيل ، والتي يمكن استخدامها للتشغيل المستمر. يمكن السيطرة على السلوك بقيادة الطرق على ()، قبالة () و فلاش (ثانية) .

 import pyb import uasyncio as asyncio class LED_async(): def __init__(self, led_no): self.led = pyb.LED(led_no) self.rate = 0 loop = asyncio.get_event_loop() loop.create_task(self.run()) async def run(self): while True: if self.rate <= 0: await asyncio.sleep_ms(200) else: self.led.toggle() await asyncio.sleep_ms(int(500 / self.rate)) def flash(self, rate): self.rate = rate def on(self): self.led.on() self.rate = 0 def off(self): self.led.off() self.rate = 0 

وتجدر الإشارة إلى أن على ()، قبالة () ، و فلاش () تمثل تقنيات متزامن التقليدية. يغيرون سلوك LED ، لكنهم يعودون على الفور. يحدث الوميض "في الخلفية". وهذا موضح بالتفصيل في القسم التالي.

يتوافق الفصل مع مبدأ OOP ، والذي يتألف من تخزين المنطق المرتبط بالجهاز في الفصل. في الوقت نفسه ، يضمن استخدام uasyncio أن يستجيب التطبيق للأحداث الأخرى أثناء وميض مؤشر LED. يومض البرنامج أدناه مع أربعة مؤشرات LED Pyboard بترددات مختلفة ، ويستجيب أيضًا لزر USR ، الذي يكمله.

 import pyb import uasyncio as asyncio from led_async import LED_async # ,   async def killer(): # ,      sw = pyb.Switch() while not sw.value(): await asyncio.sleep_ms(100) leds = [LED_async(n) for n in range(1, 4)] for n, led in enumerate(leds): led.flash(0.7 + n/4) loop = asyncio.get_event_loop() loop.run_until_complete(killer()) 

على عكس المثال الأول من حلقة الحدث ، فإن المنطق المرتبط بالمفتاح موجود في وظيفة منفصلة عن وظيفة مؤشر LED. انتبه إلى الكود المستخدم لبدء المجدول:

 loop = asyncio.get_event_loop() loop.run_until_complete(killer()) #    #       killer (), #   . 

8.4 يدعم التخطيط في uasyncio

Python 3.5 و MicroPython مفهوم الوظيفة غير المتزامنة ، والمعروف أيضًا باسم coroutine أو المهمة. يجب أن يتضمن coroutine بيان انتظار واحد على الأقل .

 async def hello(): for _ in range(10): print('Hello world.') await asyncio.sleep(1) 

تطبع هذه الوظيفة رسالة عشر مرات على فواصل زمنية مدتها ثانية واحدة. في حين أن الوظيفة متوقفة مؤقتًا تحسباً للتأخير ، فإن جدولة التزامن ستقوم بمهام أخرى ، مما يخلق وهم تشغيلها في وقت واحد.

عندما تنتظر مشكلات coroutine asyncio.sleep_ms () أو تنتظر asyncio.sleep () ، فإن المهمة الحالية متوقفة مؤقتًا وتوضع في قائمة انتظار مرتبة حسب الوقت ومتابعة التنفيذ للمهمة في أعلى قائمة الانتظار. تم تصميم قائمة الانتظار بطريقة ، حتى لو كان وضع السكون المحدد صفرًا ، سيتم تنفيذ المهام الأخرى ذات الصلة حتى يستأنف التيار. هذا هو التخطيط "دائرية صادقة". هو ممارسة شائعة لتشغيل انتظار asyncio.sleep (0) الحلقات.بحيث لا تؤخر المهمة التنفيذ. التالي عبارة عن حلقة انتظار مشغول انتظار مهمة أخرى لتعيين متغير العلم العمومي . للأسف ، يحتكر المعالج ، ويمنع إطلاق coroutines الأخرى:

 async def bad_code(): global flag while not flag: pass #  flag = False #     

المشكلة هنا هي أنه إلى أن تقوم حلقة flagis False بتمرير التحكم إلى المجدول ، فلن تبدأ أي مهمة أخرى. النهج الصحيح:

 async def good_code(): global flag while not flag: await asyncio.sleep(0) #  flag = False #     

وللسبب نفسه ، من الممارسات السيئة تعيين التأخير ، على سبيل المثال ، utime.sleep (1) لأنه يحظر المهام الأخرى لمدة ثانية واحدة ؛ من الأصح استخدام انتظار asyncio.sleep (1) .
علما بأن التأخير الناتجة عن الطرق uasyncio النوم و sleep_ms قد تتجاوز في الواقع الوقت المحدد. هذا يرجع إلى حقيقة أن المهام الأخرى سيتم تنفيذها أثناء التأخير. بعد انقضاء فترة التأخير ، لن يتم استئناف التنفيذ حتى تنتظر أو تنتهي مهمة المهمة قيد التشغيل . سوف coroutine حسن التصرف تعلن دائما تنتظرعلى فترات منتظمة. عندما يكون التأخير مطلوبًا ، لا سيما إذا كان أحدهما أقل من بضع مللي ثانية ، فقد يكون من الضروري استخدام utime.sleep_us (لنا) .

8.5 لماذا التعاونية ، وليس الجدولة القائمة على موضوع ( _thread )؟

رد فعل المبتدئين الأولي لفكرة coroutines التخطيط المشترك غالبا ما يكون مخيبا للآمال. بالتأكيد تدفق التخطيط هو أفضل؟ لماذا يجب أن أتحكم صراحةً في التحكم في الطريقة إذا كان الجهاز الظاهري Python يمكنه القيام بذلك من أجلي؟

عندما يتعلق الأمر بالأنظمة المدمجة ، فإن نموذج التعاون له ميزتان.
الأول هو الوزن الخفيف. من الممكن أن يكون هناك عدد كبير من coroutines ، لأنه على عكس الخيوط المجدولة ، تشغل coroutines المعلق مساحة أقل.
ثانياً ، يتجنب هذا بعض المشكلات الدقيقة المرتبطة بجدولة الجدولة.

في الممارسة العملية ، يتم استخدام المهام المتعددة التعاونية على نطاق واسع ، خاصة في تطبيقات واجهة المستخدم.

دفاعًا عن نموذج تخطيط البث ، سأعرض ميزة واحدة: إذا كتب شخص ما

 for x in range ( 1000000 ): #  -  

لن تمنع المهام الأخرى. يفترض نموذج التعاون أن الحلقة يجب أن تمنح كل عنصر تحكم في المهمة صراحة عددًا معينًا من التكرارات ، على سبيل المثال ، وضع التعليمات البرمجية في coroutine والإصدار الدوري في انتظار asyncio.sleep (0) .

للأسف ، هذه الميزة تتضاءل بالمقارنة مع العيوب. يتم وصف بعض هذه في وثائق لكتابة معالجات المقاطعة.. في نموذج جدولة التدفق ، يمكن لكل مؤشر ترابط مقاطعة أي مؤشر ترابط آخر ، وتغيير البيانات التي يمكن استخدامها في مؤشرات الترابط الأخرى. كقاعدة عامة ، من الأسهل بكثير العثور على القفل الذي يحدث نتيجة لخطأ لا يعطي نتيجة من اكتشاف أخطاء في بعض الأحيان ناعمة للغاية ونادراً ما تصادف والتي يمكن أن تحدث في رمز مكتوب في إطار نموذج مع تخطيط التدفق.

ببساطة ، إذا كتبت coroutine MicroPython ، يمكنك أن تتأكد من أن المتغيرات لن تتغير فجأة بواسطة coroutine آخر: coroutine الخاص بك لديه السيطرة الكاملة حتى يعود في انتظار asyncio.sleep (0) .

ضع في اعتبارك أن معالجات المقاطعة استباقية. ينطبق هذا على مقاطعات الأجهزة والبرامج التي يمكن أن تحدث في أي مكان في التعليمات البرمجية الخاصة بك.

يمكن الاطلاع هنا على مناقشة بليغة لقضايا تخطيط التدفق .

8.6 التفاعل

في التطبيقات غير التافهة ، يجب أن تتفاعل coroutines. طرق بايثون التقليدية يمكن استخدامها . يتضمن ذلك استخدام المتغيرات العامة أو تعريف coroutines كطرق كائن: يمكنهم مشاركة متغيرات المثيل. بدلاً من ذلك ، قد يتم تمرير كائن قابل للتغيير كوسيطة إلى coroutine.

يتطلب نموذج تخطيط البث المتخصصين التأكد من أن الفصول توفر اتصالاً آمنًا ؛ في نموذج التعاون ، نادرًا ما يكون ذلك مطلوبًا.

8.7. استفتاء ( الاقتراع )

بعض الأجهزة مثل جهاز التسارع Pyboard ، لا تدعم المقاطعات، ولذا ينبغي استطلاع آراء (أي فحص دوري). يمكن أيضًا استخدام الاقتراع بالاقتران مع معالجات المقاطعة: يحافظ معالج المقاطعة على المعدات ويضع علامة. يستطلع coroutine العلم - إذا تم تعيينه ، تتم معالجة البيانات وإعادة تعيين العلم. أفضل طريقة هي استخدام فئة الأحداث .