בעיות של מערכות מבו...זרות

לאחרונה שמתי לב שהנושא של "מערכות מבוזרות" הוא (שוב?) באופנה. אנשים רבים מציינים שהם מתעסקים "במערכות מבוזרת" - במה באמת מדובר? האם מדובר במערכות ווב שרבים מאיתנו עוסקים בהן? אולי רק בפרויקטים מחקריים אולטרא-מורכבים שרובנו בכלל לא נבין?

האמת... כרגיל - נמצאת איפהשהו באמצע, ובנוסף היא גם איננה חד-משמעית.

אני מניח שלאנשים שונים יש תפיסה מעט שונה לגבי מה שנכלל בתחום ה"מערכות המבוזרות" ומה המשמעות שלו.
החלטתי להקדיש פוסט קצר לעניין - כפי שאני מכיר אותו.

אם תחפשו באמזון (אני משוטט בין הספרים הטכניים שם בתור תחביב קבוע) - אין הרבה ספרות על מערכות מבוזרות. זהו תחום מצד אחד "נחשב" (נראה לי) ומצד שני שכיסוי הידע בו נמצא בחסר - מול הצורך האמיתי בשוק. יש כמה ספרים שעוסקים באלגוריתמים מבוזרים (צד אחד נקודתי של העניין) ו 3-4 ספרים שמנסים לספק תמונה מקיפה על התחום, על כלל מרכיביו - אבל הם לרוב אקדמיים למדי. לא תמצאו בהם מילה על Hadoop, קסנדרה, או Zookeeper - שהן אולי המערכות המבוזרות החשובות של תקופתנו. הם מכסים לכל היותר מערכות של תקשורת P2P.

את הרקע התאורטי שלי על מערכות מבוזרות שאבתי מספר (אקדמי, אך מהנה למדי) של אנדריי טאננבאום. זה הבחור שכנראה למדתם מהספרים שלו באוניברסיטה על מערכות הפעלה, ארכיטקטורות חומרה של מחשבים, או רשתות תקשורת.

בספר, טננבאום מגדיר מערכת מבוזרת באופן הבא:

A Collection of independent computers that appear to its users as one computer -- Andrew Tannenbaum

כאשר:

• המחשבים במערכת פועלים בו-זמנית (concurrently).
• המחשבים במערכת כושלים באופן בלתי-תלוי אחד מהשני.
• השעון של המחשבים במערכת לא מכוונים לאותה השעה.

התנאי השלישי היא המעניין: למה שלא יהיו מכוונים לאותה השעה? ובכן:

• אנו, בני-אדם, מחשיבים שעונים כמכוונים לאותה השעה גם כאשר יש ביניהם הפרש של כמה שניות [א]. עבור מחשבים, הפרש של כמה מליוניות שניה - יכול לעשות את ההבדל.
• יש להניח שהתקשורת בין המחשבים במערכת עוברת ברשת - רשת בה יש אקראיות. אפילו אם המחשבים קרובים זה לזה פיסית, הרשת תגרום לזה שהודעה שנשלחה בזמן t לשני מחשבים שונים תגיע למחשבי היעד בזמנים שונים ובסדר אקראי - אקראיות שניתן להקביל, ל "שעונים שלא מכוונים אותו הדבר".

כלומר: אם נניח מראש שהשעונים אינם מכוונים, הסיכוי שלנו להיכשל בהנחות שגויות לגבי התנהגות המערכת - יפחת, ולכן אולי כדאי לנו להניח שזה המצב הנתון.

מחשב אישי (PC) הוא לא מערכת מבוזרת בדיוק מהסיבה הזו.
חשבו על זה: המחשב האישי בנוי מכמה רכיבים דיי חכמים ("מחשבים"?), שפועלים במקביל - ויכולים להיכשל באופן בלתי תלוי: CPU, GPU (מעבד או "כרטיס" גראפי), כונן כשיח, כרטיס רשת, וכו'.

האם המחשב האישי הוא מערכת מבוזרת?

לא. הסיבה לכך היא שהמעבד הראשי (CPU) הוא הרכיב שמכתיב את קצב העבודה ("מספק שעון", מאות אלפי פעמים בשנייה) לכל הרכיבים האחרים במערכת - וכל הרכיבים הם מסונכרנים בדיוק לאותו השעון. ההבדל סמנטי, לכאורה - אבל חשוב.


הערה חשובה: מערכות מבוזרות הן מורכבות יותר ממערכות שאינן מבוזרות. אם תצליחו לפתור בעיה נתונה בעזרת מערכת לא-מבוזרת - כנראה שזה יהיה פתרון טוב יותר.

בעיות "אקדמיות" של מערכות מבוזרות 

אלו לרוב בעיות שהן יותר low level, בליבה של המערכת המבוזרת - שאולי רבים מהמשתמשים של המערכות הללו יכולים לא להכיר.


Multicast - שליחת הודעה לכל המחשבים בקבוצה

אמנם פרוטוקול IP כולל יכולות Multi-cast כחלק מהפרוטוקול - אבל: א. המחיר של multicast הוא גבוה למדי, ב. ברוב הפעמים קונפיגורציה הרשת (הפרדה בין רשתות, firewalls, וכו') והעובדה שחלק מהמכונות לא זמין לקבל את ההודעה באותו הרגע (בשל כשל / עומס) - הופך multicast ברמת הרשת לכמעט לא רלוונטי.

את בעיית ה Multicast פותרים לרוב ב-2 דרכים עיקריות:

• Messaging - כאשר יש מתווך (Message Bus, Message Broker וכו') שדואג לכך שההודעות יגיעו גם למחשב שכרגע לא זמין. המתווך צריך להיות Highly Available ובעל יכולת גישה לכל המחשבים במערכת - על מנת לספק רמת שירות גבוהה.
• Gossip-based transmitting (נקרא גם Gossip Protocol) - משפחה של פרוטוקולים שמעבירים את ההודעה כחיקוי הדפוס של התפשטות מגיפות (אפידמיה): כל מחשב שולח הודעה לחבר אקראי, אחת לכמה זמן, לאורך טווח זמן שנקבע. סטטיסטית, בסבירות גבוהה מאוד - ניתן להגדיר התנהגות שבסופה כל המחשבים ברשת יקבלו לבסוף את ההודעה, על אף כשלים, שינוי בתוואי הרשת, או קושי להגיע למחשבים מסוימים. החיסרון של גישה זו היא שטווח הזמן של פעפוע ההודעות איננו מהיר, ויש "בזבוז" של משאבי הרשת בהעברת הודעות כפולות. לרוב משתמשים בדפוס זה להעברת הודעות קטנות.
  יישומים מקובלים הם עדכון קונפיגורציה או איתור ועדכון על כשלים במערכת.

Remote Procedure Call (בקיצור RPC)

הבעיה של הפעלת מתודה על מחשב מרוחק, בצורה אמינה וקלה - גם היא נחשבת לסוג של בעיה של מערכות מבוזרות, אבל מכיוון שהדומיין הזה מפותח למדי - לא אכנס להסברים.

Naming - היכולת של כל המחשבים במערכת לתת שם אחיד למשאב מסוים

דוגמה נפוצה אחת יכולה להיות הכרת כל המחשבים במערכת (למשל hostnames), בעוד מחשבים עולים ונופלים כל הזמן.

דוגמה נפוצה אחרת היא כאשר נתונים זזים לאורך הזמן (בשל רפליקציה, כשלים, ו partitioning) בין אמצעי אכסון שונים - ואנו רוצים להיות מסוגלים לאתר אותם.

יש גישות רבות לפתרון הבעיה, אציין כמה מהן בזריזות:

• Naming Server מרכזי (או מבוזר עם רפליקציות לקריאה-בלבד, דמוי DNS -שהוא גם היררכי) - שמנהל את השמות במערכת וכולם פונים אליו.
• Client Side Directory (קליינט = מחשב במערכת) - גישה בה משכפלים את ה Directory לכל המחשבים במערכת ע"י multi-cast (למשל עבור client-side load balancing - כאשר כל מחשב ברשת בוחר אקראית למי לפנות בכדי לבקש שירות. סוג של פתרון לבעיית ה Fault Tolerance).
• Home-Based Approaches - כאשר הקונפיגורציה היא כבדה מדי מכדי לשכפל כל הזמן בין כל המחשבים במערכת, ניתן לשכפל (ב multi-cast) רק את רשימת שרתי הקונפיגורציה (עם אורנטצייה גאוגרפית, או של קרבה - ומכאן המונח "Home"). אם שרת קונפיגורציה אחד לא זמין (כשל בשרת או בתוואי הרשת) - ניתן לפנות לשרת קונפיגורציה חלופי.
• Distributed Hash Table (בקיצור DHT) - שזה בעצם השם האקדמי למבנה הנתונים הבסיסי של שרת Naming משתכפל עצמית - לכמה עותקים, בצורה אמינה, scalable, וכו'. בסיסי-נתונים מבוזרים מסוג K/V כמו Cassandra או Riak - נחשבים למשתייכים לקטגוריה זו, ויכולים (ואכן משמשים) כפתרון לבעיית Naming במערכת מבוזרת.
  למיטב הבנתי, Cassandra מתבססת על מבנה מעט שונה שנקרא Consistent Hashing, שיעיל ביותר בפעולות ה Lookup של keys, על חשבון מחיר שינויים ברפליקציה שהם יקרים יותר. הנה מצגת שמצאתי שמסבירה את ההבדל בין מבני-נתונים.

סנכרון שעונים 

אמנם אמרתי בהקדמה שהנחת היסוד של מערכות מבוזרות היא שהשעונים אינם מסונכרנים - אך זה לא אומר שלא ניתן לשאוף למצב כזה, או לקירוב שלו. העניין הוא גם שכל המחשבים במערכת יגיעו להסכמה על שעון אחיד (בעיית הקונצנזוס - נדון בה מיד) וגם היכולת לקחת בחשבון את ה Latency של הרשת ו"לבטל אותו", כלומר: בקירוב מסוים. 

לעתים קרובות, הידיעה שלמחשבים במערכת יש שעות אחיד בסטייה ידועה (נניח עד 100ns, בסבירות של 99.99%) - יכול להיות בסיס לפישוט תהליכים משמעותי. גישות להתמודדות בעיה זו:

• הפתרון הבסיסי ביותר הוא פתרון כמו Network Time Protocol (בקיצור NTP) שמקובל על מערכות לינוקס / יוניקס - שקורא את השעה משרת זמן מרכזי, ותוך כדי הקריאה מבצע מדידות של ה latency לשרת המרכזי. באופן זה, הוא יכול להסתנכרן, בסטייה שניתנת לחישוב - לזמן של השרת המרכזי.
• פרוטוקולים משוכללים יותר, כמו Reference broadcast synchronization (בקיצור RBS) אינם מניחים על קיומו של שרת זמן מרכזי, אלא מודדים latencies לשאר המחשבים במערכת - וכך מייצרים זמן משוערך יחסית למחשבים אחרים במערכת.
• חשוב לציין שהשעון לא חייב להיות שעון של בני-אדם (שעה, דקה, שנייה, מיקרו שנייה), אלא יכול להיות שעון לוגי, שהוא בעצם counter עולה של אירועים. כל פעם שיש אירוע בעל משמעות - מעלים את ה counter באחד ומסנכרים את כלל המערכת על רצף האירועים (ולאו דווקא זמן ההתרחשות המדויק שלהם). מימוש מקובל: Lamport's Clock.
  הרעיון דומה לרעיון ה Captain's log stardate הדמיוני מסדרת המדע הבדיוני StarTrek: מכיוון שמסע במהירות האור (או קרוב לו) משפיע על מרחב הזמן, לא ניתן להסתמך על שעון עקבי עם שאר היקום, ולכן הקפטן מנהל יומן ע"פ תאריכים לוגיים. (סיבה נוספת למנגנון התאריכים בסדרה: לנתק את הצופה מרצף הזמן המוכר לו, סיבה קולנועית גרידא).

יצירת קונצנזוס

בעיית הקונצנזוס היא הבעיה לייצר תמונת עולם אחידה, לגבי פריט מידע כלשהו - עבור כל המחשבים במערכת.

ברמה הבסיסית יש את בעיית ה Leader Election: כיצד בוחרים מחשב יחיד בקבוצה (נקרא "מנהיג" - אבל עושה את העבודה השחורה) לבצע פעולה כלשהי (למשל: לשלוח מייל למשתמש, על אירוע שכל המחשבים במערכת יודעים עליו).
אנו רוצים שרק מחשב יחיד יבצע את הפעולה בכדי למנוע כפילויות.
כיצד בוחרים מנהיג?

• לנסות לדמות בחירות אנושיות - זה לא כיוון פעולה מוצלח. נדלג. :)
• אלגוריתם פשוט (The Bully Algorithm) מציע לכל מחשב במערכת לשלוח ב broadcast מספר אקראי שהגדיר, ובעל המספר הגבוה ביותר - הוא המנהיג.
• ישנם מספר אלגוריתמים מורכבים יותר (מבוססי Ring או DHT) - שהם אמינים יותר, ומתאימים יותר למערכות עם מחשבים רבים.
• בכל מקרה, אחד העקרונות המקובלים הוא לבחור מנהיג קבוע (כלומר: עד שהוא מת) - בכדי לחסוך את הפעלת האלגוריתם שוב ושוב. זה יכול להיות "מנהיג אזורי" או "מנהיג גלובאלי". 

סוג אחר יש יצירת קנוזנזוס נעשית ע"י טרנזקציה מבוזרת. בד"כ מבצעים טרנזקציה מבוזרת ע"י רעיון / אלגוריתם שנקרא two-phase commit (או בקיצור 2PC). האלגוריתם עובד כך:

1. ה coordinator הוא מי שיוזם את הפעולה. הוא שולח vote request לכל המחשבים המיועדים להשתתף בטרנזקציה המבוזרת.
2. כל מחשב שמקבל את ה vote request עונה vote-commit (אם הוא מסוגל לבצע את הפעולה) או vote-abort (אם הוא לא מסוגל לבצע אותה).
3. ה coordinator אוסף את כל הקולות. אם יש לפחות מישהו אחד שענה vote-abort, או פשוט לא ענה - הוא מפרסם global-abort וכל המחשבים מבצעים rollback.
4. אם כולם ענו ב vote-commit - ה coordinator שולח הודעת global-commit - וכל המחשבים מבצעים commit מקומי על הטרנזצקיה.

יש גם גרסה מתוחכמת יותר של הפרוטוקול, בשם three-phase-commit (בקיצור 3PC) - המתמודדת עם מצב ביניים בו ה coordinator כשל. מעולם לא שמעתי מערכת שממשת את 3PC, וכנראה ש 2PC טוב לרוב הגדול של השימושים.

עוד אלגוריתם מקובל להשגת קונצנזוס הוא paxos שדורש שרוב (quorum) של המחשבים במערכת יסכים על עובדה - ואז הם מעדכנים את השאר המחשבים על ההחלטה.

סוג אחרון של קונזנזוס, או תאום הוא Distributed Mutual Exclusion (שנקרא בתעשייה פשוט Distributed Lock) בו כמה מחשבים מתאימים ביניהם מי ייגש למשאב מסוים בלעדית, ייתכן ובמצב שלמשאב הזה יש כמה עותקים במערכת (בשל רפליקציה). 

Distributed Fault Tolerance

הכלל השחוק כמעט בעניין זה הוא "Avoid a single point of failure" - על מנת שמערכת מבוצרת תהיה אמינה, יש לדאוג שאין שום רכיב יחיד במערכת שכישלון שלו - מכשיל את כלל המערכת.

תחום זה כולל נושאים של:

• גילוי כשלים במערכת. אם מחשב א' לא מצליח ליצור קשר עם מחשב ב' - האם זו בעיה במחשב א', במחשב ב'? או בשילוב שלהם (גרסאות לא מתואמות, בעיות רשת)?
• הבסיס לפתרונות הוא לרוב בעזרת coordinator (או קבוצה של coordinators) שדוגמים את כלל המחשבים במערכת בעזרת הודעות heartbeat (או alive) - בהן כל מחשב מדווח על מצבו, וע"פ כללים מסוימים מגיעים להחלטה.
• גישות מתוחכמות יותר מתבססות על מידע שמגיע ממחשבים על חברים שלהם במערכת - ואז לקיחת החלטה מבוזרת (למשל: הצבעה בנוסח paxos האם מחשב מסוים הוא תקין או לא).
• תקשורת P2P שהיא כבר פתרון לקשיים בין מחשבים במערכת לתקשר זה עם זה. כאשר מתקינים מערכת ברשת ארגונית - ייתכנו מצבים (ע"פ הגדרה) בהם אין לכל המחשבים ברשת תקשורת ישירה זה עם זה. מצב בעייתי אחר שאיתו P2P מתמודד הוא כשלים ברשת האינטרנט (ויעילות בהעברת כמויות גדולות של מידע, למשל: סרטים).
• נושא נוסף הוא אלגוריתמים מתייצבים עצמית, שזה הרעיון של כתיבת אלגוריתמים שבהינתן מצב לא תקין רנדומלי באחד המחשבים - האלגוריתם, תוך כדי פעולה, יפצה על הכשל ויתקן אותו עם הזמן. פרוטוקולי תקשורת, ופרוטוקולים של רפליקציית נתונים - כוללים אלמנטים של התייצבות-עצמית.

כיום, ניתן למצוא כמה מערכות שמספקות את ה primitives הנ"ל. מערכות כמו:

• Zookeeper (של Hadoop)
• Consul (מבית היוצר של Vagrant)
• Etcd
• Eureka (של נטפליקס)

מספקות בבסיסן מערכת Naming מבוזרת (הם קוראים לזה לרוב Service Discovery) וקונפיגורציה מבוזרת (משהו בין Naming מבוזר לניהול קונצנזוס), אבל חלקן כוללות גם פרימיטיביים של Leader Election, Voting, 2 Phase-Commit וכו'.

אם אתם מזהים במערכת שלכם את הצורך באחד מהפרימיטיביים הללו - שווה אפשר לשקול לקחת מערכת מוכנה.

האמת: פעם בחיים פיתחנו Leader Election, ופעם 2PC - ודווקא היה בסדר (בהקשר לפוסט "התשתית הטובה ביותר").

בעיות "מעשיות" של מערכות מבוזרות

כאשר ניגשים למימוש מערכות מבוזרות "רגילות" (למשל: מערכות ווב ב Scale), אנו כנראה ניתקל בכמה בעיות שונות מהבעיות הנ"ל: בעיות ברמת הפשטה גבוהה יותר - שניתן לפתור בעזרת כלים שהתמודדו בעצמם עם חלק מהבעיות הנ"ל.

רק בשלבים מאוחרים יותר של ההתמודדות עם המערכת, אנו נדרשים ל fine-tuning שחושף אותנו להתמודדות עם הבעיות "האקדמיות" שהצגתי.


הבעיות המעשיות עליהן אני מדבר הן:

בעיות Scale

בעצם הבעיה שהופכת את המערכת שלנו למערכת מבוזרת - והיא עושה זאת לרוב בשלבים דיי מוקדמים. כאשר שרת אחד (גדול?) לא מספיק לבצע את המשימה, יש כמה גישות עיקריות לפעולה:

1. שכפול "כוח-עבודה זול" - מעבר ל cluster של מחשבים שיבצע את העבודה (דאא). מה שנקרא "ציר-x" של ה Scale Cube.
2. העברת עבודה ל Clients (למשל קוד javaScript). מכיוון שעל כל שרת יש עשרות עד אלפי מחשבי Client - יש הגיון רב לנצל את כח החישוב שלהם או האכסון שלהם.
3. יצירת עותקים נוספים לקריאה-בלבד (Read Replicas) כפי שנהוג ב MySQL, למשל.
1. וריאציה אחרת של רעיון זה הוא שימוש ב caches.
4. פירוק המערכת למספר תתי-מערכות שכל אחת מבצעת תת-פונקציה. מה שנקרא "ציר ה-y" של ה Scale cube. חלוקת מערכת "מונוליטית" למיקרו-שירותים היא סוג של פירוק מערכת לתתי-מערכות.
5. ביצוע Partioning (של מידע או עבודת חישוב). למשל: מחשבים בקבוצה א' יעבדו את הלקוחות ששמם מתחיל באותיות א-י, והמחשבים בקבוצה ב' יעבדו את הלקוחות ששמת מתחיל באות כ-ת. חלוקה זו מסומנת כ "ציר-z" ב Scale Cube, ובד"כ מגיעים אליה לאחר ששאר המאמצים מוצו.
   שימוש ב Partitioning מקל על ה Scalability, אבל יוצר בעיות חדשות של Consistency וסנכרון.

ה Scale Cube הוא מודל תאורטי, אך שימושי - המאפיין את הדרכים להגדיל את אופן פעולתה של מערכת (בעיקר צד-שרת).
בד"כ מתחילים לגדול בציר x, לאחר מכן ציר y, ורק לבסוף בציר z (המפרך יותר).

כאשר עוסקים ב Scale גדול המעבר ל Partitioning הוא כנראה בלתי-נמנע, מה שמציב כמה בעיות אחרות.
בעיה אחת לדוגמה היא הבעיה הידועה כ CAP theorem. הטענה שבהינתן Partitioning, לא ניתן לקבל גם Availability (זמינות) ו Consistency (עקביות הנתונים) באותו הזמן - ויש לבחור ביניהם.

או שנתשאל את המערכת ותמיד נקבל תשובה (Availability גבוה) - אבל התשובה לא בהכרח תהיה זהה מכל המחשבים שנפנה אליהם - כלומר: ביצענו פשרה ב Consistency, מה שנקרא AP.
או שנתשאל את המערכת ותמיד נקבל תשובה זהה מכל המחשבים במערכת (Consistency גבוה) - אבל לפעמים לא תהיה תשובה, כלומר התשובה תהיה "לא בטוחים עדיין, אנא המתן) (פשרה ב Availability), מה שנקרא CP.

הערה1: בסיס נתונים רלציוני נחשב כ "CA" כי הבחירה שלו היא ב Consistency ו Availability - אך ללא יכולת ל Partitioning (ולכן אומרים ש RDBMs הוא "לא Sacalable").
הערה2: פרויקט Cassandra הצליח לכאורה "לרמות" את ה CAP Theorem. כיצד? הנה הסבר מהיר: כל פריט מידע נשמר (נניח) כ 5 פעמים במערכת. בכל פעולת קריאה אנו מציינים כמה עותקים נדרשים לצורך הקריאה. אם נבחר 1 - נקבל AP, אם נבחר 5 - נקבל CP - ויש לנו את הטווח באמצע. כלומר: לא באמת רימינו את ה CAP Theorem, אבל אפשרנו לכל פעולת קריאה לבצע את ה trafeoff עבור עצמה.

עוד הקבלה מעניינת של ה CAP Theorem נמצאת בפעולת multi-casting השונות:

פרוטוקול Gossip הוא סוג של AP, פרוטוקול Paxos הוא סוג של CP, ופרוטוקול 2PC הוא סוג של CA - הוא לא יכול להתבצע על חלק מהמחשבים במערכת ולהשיג את התוצאה.


עוד עניינים מעשיים, שכתבתי עליהם בפוסטים קודמים:

• Circuit Breakers - ואיך לשמור על האמינות של המערכת (צמצום נקודות Failure).
• כיצד לחלק מערכת לשירותים מבוזרים. פוסט ישן, עדיין רלוונטי - אך הייתי נותן היום משקל גדול יותר לענייני אופרציה / DevOps.
• ארכיטקטורת מיקרו-שירותים - ממד אחד ומקיף יותר, של השאלה איך לחלק מערכת לשירותים מבוזרים.

סיכום

פוו... זה היה כיף! ישבתי עכשיו איזה חמש שעות רצופות וכתבתי. עוד מעט שלוש בבוקר.
אני מקווה שבפוסט הצלחתי לסדר קצת יותר טוב בראש את היקף התחום של "מערכות מבוזרות", לספק תמונה כללית על הבעיות העיקריות - והאמצעים לפתרון, ואולי גם להפיג אולי חלק מ"ערפל המיתוס" מסביב לתחום - לאלו שלא מכירים אותו. בסופו של דבר "מערכות מבוזרות" הוא עוד תחום של הנדסת תוכנה.

שיהיה בהצלחה!


----

קישורים רלוונטיים

קישור לחומרי קורס "מערכות מבוזרות" באוניברסיטת אמסטרדם, שנראה טוב למדי. מצגות, וידאו.


-----

[א] אני מכיר דווקא בחור אחד שלא. אצלו "מכוונים לאותה השעה" הוא הפרש של עד חצי שנייה - ואל תשאלו איך הוא מכוון אותם....