מבנה של pipeline

[nocgod]

מבנה של pipeline

אני עובד על פרוייקט שמנתח טקסטים ובסופו של דבר נותן תוצר.
הרגע הרעיון הכללי הוא: קרא טקסטים -> נתח טקסטים -> בנה מודלים -> חשב תוצר
הרעיון הוא שהכל יהיה מקבילי, כלומר אני לא אחכה לקריאת כל ה n טקסטים שיש לי במאגר, אלא ברגע שיש לי טקסט שקראתי אני אעביר אותו הלאה.
אני קורא בימים האחרונים המון על pipelines ו actor model, ואני עדיין לא מצליח לחבר לעצמי תמונה נורמלית של איזה זה אמור להיות.
אני מגדיר קריאה, ניתוח, בנייה וחישוב כשלבים (כל אחד שלב נפרד), ואני רוצה בכל שלב לבצע את העבודה על כמה threads במקביל.

יש למישהו מכם רעיון על framework/library מוכן שיכול לעזור לי עם העבודה של הסנכרון?
נראה לי תמוהה ומוזר שאני לא מצליח למצוא ספריה שתיתן לי backend שיש בו את כל הסנכרון והmessage passing וכל מה שאני צריך לספק לה זה את הstage implementation ואת ה stageWorker implementation.
יש מצב אני מחפש לא נכון...

תודה מראש חברים...

 

[selalerer]

באיזו שפה?

 

[nocgod]

סליחה שכחתי לציין - Java

רציתי להשתמש בzeroMQ כbackend שלי אבל זה סיבוך רק לקמפל אותו לwindows והמחשבה שהמערכת צריכה להיות cross platform אומרת שאני צריך לעבור את הסיבוך הזה עבור כל גרסא עבור כל מערכת דיי מוריד לי מזה...
אני מציץ עכשיו בAKKA שיהיה הbackend שלי להעברת הודעות בין שלבים, זה עדיין לא משנה את העבודה שהמימוש של הpipeline הנוכחי שלי צריך לאבד (יש לי שיער לבן בגלל זה) מידע בדרך

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

 

[nocgod]

הספרייה שלי

הרגע (אחרי יומיים מתישים של חיפוש race conditions) זה עובד.
https://bitbucket.org/nocgod/parallel-pipeline-processing-library
אני בדקתי את העבודה של זה על פרוייקט שלי וקיבלתי תוצאות כצפוי. אני מאמין שיש לאן לפתח עוד את הספריה ואשמח לשיתוף פעולה...
מי שרוצה להיות contributor שיפתח חשבון bitbucket וישלח לי מייל אני אפתח גישה לפרוייקט.

 

[Javali]

בגים פוטנציאליים

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

ככלל העלות של synchronized כאשר אין התנגשות זניחה. כאשר יש התנגשות, העלות היא בלתי נמנעת.

 

[nocgod]

מקובל...

למה שלא תפתח issue בביטבאקט בנוסף? מצידי אפילו אל תתרום קוד, רק issues זה גם עזרה...
הissue tracker ציבורי וכל אחד יכול לפתוח issue...
אם תרצה לעזור עם קוד אני אוסיף אותך...בכיף

 

[selalerer]

הקוד שלך מאוד מסובך שלא לצורך.

אם הבנתי נכון מה אתה מנסה לעשות, אתה בסך הכל צריך queue לכל stage. ליצור כמה threadים שאתה רוצה עבור כל stage שיחכו על ה-queue ויעבירו את מה שמתקבל לאובייקט שמבצע את הלוגיקה של ה-stage. זה יחזיר בסוף הלוגיקה message אחד או יותר שילך לתוך ה-queue של ה-stage הבא.

אם יש כמה stageים שיכולים לחלוק את אותם ה-threadים (למשל העבודה שלהם היא CPU bound) אז תן להם את אותו queue ולפי סוג ה-message תעשה לו dispatch אל הלוגיקה המתאימה לטיפול בו.

ב-BaseStage יש לך שם מליון memberים שלא כלכך ברור מה תפקידם. יש לך פה פרוייקט קטנטן וכבר הספקת לעשות קובץ עם 500 שורות וזה בלי שום inner class, רק class אחד. מיותר לחלוטין. גם כל הקוד הזה נחוץ, ואני בספק שהוא נחוץ, צריך לחלק אותו לכמה וכמה מחלקות קטנות יותר.

 

[nocgod]

אני אשמח

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

הספרייה מפעילה stages על threads נפרדים, מאחר וכל stage יכול להיות blocking או non-blocking כלומר המידע ממנו יוצא ברגע שהוא מוכן או ברגע שכל יחידות העבודה מוכנות.
הבעיה עם זה שstage יכול להיות חוסם או לא חוסם היא שסטייג' יכול להרים את הthreads שלו ואז לברוח החוצה מבלי לחכות לסיום, וזה כמובן רע.
הרבה קוד שם גם בנוי על מנת לעשות אבסטרקציה, משתמש של הספריה רק צריך לספר לוגיקה של worker ואם צריך אז staging של stage. עד כדי כך פשוט.

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

אגב לא שופטים קוד לפי כמות השורות (אני מניח שהסתכלת על base stage ראית שיש שם 500 שורות כשבפועל רק הדוקומנטציה הקיימת לוקחת לפחות 200-300 שורות).
הרבה אנשים שפטו קוד שלי לפי האורך שלו, ובסופו של יום הקוד שלי היה הרבה יותר דינאמי וניתן לשינוי - ובכך יותר שימושי.

אם תרצה שלח לי מייל או פרסם את המייל שלך כאן ואני אתן לך גישה לספריה שתוכל לשנות ולהוסיף קוד...אני אשמח.
אין הרבה ספריות שעושות את מה שאני מנסה לעשות בצורה פשוטה, ומה שקיים או שהוא overkill או שהוא מת כמו apache-common-pipeline או שהוא underdocumented כמו appengine-pipeline.