האזנה לערוץ בתוך כמה פונקציות במקביל

[nigun]

אני מנסה לבנות אפליקציה שתפעיל הרבה תהליכים במקביל שתאזין לערוץ מסיום וכשמתקבל ערך בערוץ כולם ימשיכו הלאה
אני ינסה להסביר את זה בקצת קוד(golang(

package main

import (
    "fmt"
    "time"
)

func main() {
	c := make(chan string)

	go func(c chan string) {
	    time.Sleep(2 * time.Second)
	c <- "foo"

	}(c)
	
	go func() {
	s:= <-c
	   		fmt.Println(s )

	}()
	go func() {
	s:= <-c
	   		fmt.Println(s )

	}()
	
	    time.Sleep(4 * time.Second)

}

כאן אני יוצר בבת אחת 3 פונצקציות במקביל
ושתי האחרונות מחכות שיתקבל מידע בערוץ
אחרי 2 שניות מתקבל מידע בערוץ
ואחד הפונקציות מקבלת את המידע וממשיכה הלאה
אבל כיוון שהמידע נמחק מהערוץ הפונקציה השניה ממשיכה לחכות

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

(אפשר להשתמש בלולאת FOR שתרוץ לעולם עד שהמשתנה יהיה שווה לערך שאני רוצה (עם משתנה רגיל)
השאלה האם זה לא מידי בזבזני במשאבים)

[yossiz]

@nigun תודה על השאלות בזכותך יוצא לי ללמוד קצת GO...

(אפשר להשתמש בלולאת FOR שתרוץ לעולם עד שהמשתנה יהיה שווה לערך שאני רוצה (עם משתנה רגיל)
השאלה האם זה לא מידי בזבזני במשאבים)

ברור שזו לא הדרך.

משיטוט קצת ברשת מצאתי 3 פתרונות (קח הכל בעירבון מוגבל).

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

מקורות: 1, 2

[nigun]

@yossiz אמר בהאזנה לערוץ בתוך כמה פונקציות במקביל:

• איתות באמצעות סגירת ערוץ שזה אירוע שמשודר לכל המאזינים

לא עוזר לי, כי אני רוצה לעשות שימוש בערך אצל כל המאזינים.

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

נראה לי שזה עוזר רק אם ידוע מראש מספר המאזינים (במקרה שלי אני לא יודע מי וכמה יהיו המאזינים מראש)

• שימוש ב-Cond שממש מיועד לזה

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

[yossiz]

@nigun אמר בהאזנה לערוץ בתוך כמה פונקציות במקביל:

נראה לי שזה עוזר רק אם ידוע מראש מספר המאזינים (במקרה שלי אני לא יודע מי וכמה יהיו המאזינים מראש)

עיין כאן בקטע: "When the number of listeners is not known"

[nigun]

@yossiz
קצת קיבלתי סחרחורת מכל הערוצים
לא הצלחתי עדיין ליצור לעצמי דוגמא קטנה שעובדת
יש לך הפניה לדוגמה פשוטה להבנה?

[יוסף בן שמעון]

@nigun יתכן שהמקרה שלך שונה מהמקרה של השואל בסטאק, הוא היה צריך לשלוח כמה פעמים מידע לאותו ערוץ, מהדוגמת קוד שלך נראה שכל המאזינים הם חד פעמיים, אם זה המצב, אתה יכול להסתפק במערך שמחזיק את כל המאזינים, וכשהערוץ הראשי שולח מידע אתה עובר בלולאה על המערך ומעביר את המידע לכלל המאזינים.
https://play.golang.org/p/-ynOQTKzXMu

[nigun]

@יוסף-בן-שמעון
אני צריך לשלוח מידע כמה פעמים לאותם ערוצים (עם שרת HTTP)
הפיתרון שהובא לעיל נראה לי מציון רק אני צריל ללמוד איך הוא עובד בדיוק
כדי ליישם אותו בפועל

[יוסף בן שמעון]

@nigun הקוד שהבאתי הוא בסיס מופשט להבנת הענין

package main

import (
	"fmt"
	"time"
)

var (
	// הערוץ הראשי שאליו מוזרם המידע
	mainChan = make(chan string)
	// מערך של ערוצי משנה
	// כל פונקציה שתרצה לקבל מידע מהערוץ הראשי תצטרך ליצור ערוץ משנה ולהוסיף אותו למערך הזה
	channels = []chan string{}
)

func main() {

	go func() {
		// פונקציה היחידה שמקבל את המידע מהערוץ הראשי ומזרימה אותו לערוצי המשנה
		msg := <-mainChan
		for _, c := range channels {
			c <- msg
		}
	}()

	go func(c chan string) {
		time.Sleep(2 * time.Second)
		c <- "foo"
	}(mainChan)

	go someThread()
	go someThread()
	go someThread()
	go someThread()

	time.Sleep(4 * time.Second)

}

func someThread() {
	c := addListener()
	s := <-c
	fmt.Println(s)
}
func addListener() chan string {
	// הפונקציה שתפקידה ליצור את ערוצי המשנה ולהוסיף אותם למערך כדי להאזין למידע מהערוץ הראשי
	newChan := make(chan string)
	channels = append(channels, newChan)
	return newChan
}

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

package main

import (
	"fmt"
	"time"
)

var (
	mainChan = make(chan string)
	channels = []chan string{}
)

func main() {

	go func() {
		for { // << לולאה אינסופית שמאזינה לערוץ הראשי
			msg := <-mainChan
			for _, c := range channels {
				c <- msg
			}
		}
	}()

	go func(c chan string) {
		time.Sleep(1 * time.Second)
		c <- "foo"
		time.Sleep(1 * time.Second)
		c <- "foo" // << שליחת מידע לערוץ הראשי פעם נוספת
	}(mainChan)

	go someThread()
	go someThread()
	go someThread()
	go someThread()

	time.Sleep(4 * time.Second)

}

func someThread() {
	c := addListener()
	for { // << לולאה אינסופית שמאזינה לערוץ המשנה
		s := <-c
		fmt.Println(s)
	}
}
func addListener() chan string {
	newChan := make(chan string)
	channels = append(channels, newChan)
	return newChan
}

עד כאן הכל פשוט, הבעיה מתחילה בסיטואציה של השואל בסטאק, שיתכנו מאזינים שנרשמו לערוצי המשנה, ובהמשך הם יתנתקו וירדו מההאזנה, הבעיה היא שכל מאזין מריץ לולאה אינסופית, ואם לא תעצור את הלולאה כשהוא יתנתק זה יגרום לשימוש מיותר בזיכרון. בשביל זה הוא צריך להוסיף לכל מאזין גם תנאי עצירה, ולכן הוא הוסיף את הערוץ quit, וזה סיבך את הקוד.
האם אתה גם צריך את זה? זה תלוי באופי השימוש של המאזינים שאתה מקים, אם הם אמורים לרוץ כל חיי האפליקציה, אז זה מיותר, אם יש סיכוי שחלק מהם או כולם יגמרו את התפקיד שלהם וימותו, אז תצטרך לדאוג לעצירת הלולאה.
גם למקרה שצריך לעצור את הלולאה, אם המקרה של סיום התפקיד של הפונקציה ידוע וצפוי מראש, לא תצטרך להשתמש בשיטה של הערוץ quit, אלא פשוט תעצור את הלולאה בעצמך, אם זה לא תלוי בידך אלא זה מקרה לא צפוי (כמו במקרה של השואל שם שמשתמש יתנתק) לזה נועד השימוש בערוץ וב select

---

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

[nigun]

@יוסף-בן-שמעון
עזרת לי מאוד
אבל נראה לי שיש כאן בעיה
כי אם אני מוסיף עוד מאזינים אחרי הקריאה הראשונה
הם לא מצליחים להאזין
משהו כזה:

        go someThread()
        go someThread()
        time.Sleep(1 * time.Second)
     //קריאה ראשונה לערוץ הראשי 
  func(c chan string) {
                c <- "foo"
        }(mainChan)
        time.Sleep(1 * time.Second)
// יצירת מאזנים חדשים
        go someThread()
        go someThread()
// קריאה שניה לערוץ הראשי
        func(c chan string) {
                c <- "foo2"
        }(mainChan)

        time.Sleep(4 * time.Second)
}
func someThread() {
        c := addListener()
        //for { // שים לב שכאן אני מאזין רק פעם אחת לערוץ
        s := <-c
        fmt.Println(s)
        //      }
}

נראה לי שאחרי שנעשה שימוש במערך של הערוצים משום מה אי אפשר להוסיף לו עוד אברים.

[יוסף בן שמעון]

@nigun הקוד הזה עובד מצוין
https://play.golang.org/p/YOYhQerLKFd
כמובן שאחרי המידע הראשון שנשלח לערוץ המאזין נסגר ואי אפשר לשלוח עוד מידע, אבל המידע הראשון מתקבל גם לפונקציות המאוחרות.
הקוד שהעלת קטוע, אולי תעלה דוגמת קוד שלימה עם בעיה

[יוסף בן שמעון]

@nigun טעיתי, בקוד הזה באמת שליחת המידע מתבצעת רק אחרי שכל המאזינים נרשמים, אם נעשה את זה כך:
https://play.golang.org/p/hjKVDAq2hLT
לא נראה את המידע השני, אבל זה לא בגלל שאי אפשר להוסיף איברים למערך, אלא בגלל שבזמן שאמור להישלח המידע הקומפיילר עדיין באמצע השינה, וכשהוא מתעורר מהשינה כבר מאוחר כי הפונקציה main כבר מתה.
ניסיתי לעשות את זה עם שרת HTTP וזה נראה עובד בסדר גמור

package main

import (
	"fmt"
	"net/http"
)

var (
	mainChan = make(chan string)
	channels = []chan string{}
)

func main() {

	go func() {
		for {
			msg := <-mainChan
			for _, c := range channels {
				c <- msg
			}
		}
	}()

	http.HandleFunc("/a", sendDataToChannel)
	http.HandleFunc("/b", someThread)
	http.ListenAndServe(":8090", nil)

}

func sendDataToChannel(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "")
	mainChan <- "foo"
}
func someThread(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "")
	go func() {
		c := addListener()
		// for {
		s := <-c
		fmt.Println(s)

		// }
	}()
}

func addListener() chan string {
	newChan := make(chan string)
	channels = append(channels, newChan)
	return newChan
}

[nigun]

@יוסף-בן-שמעון
זה לא עובד לי
אני נכנס קודם לכתובת
http://127.0.0.1:8090/b
פעמיים
ואז פעם אחת ל
http://127.0.0.1:8090/a
ואני מקבל את הפלט (בטרמינל)
foo
foo
עד כאן הכל טוב ויפה
אבל אם אני מנסה להיכנס שוב ל
http://127.0.0.1:8090/b
ואז שוב ל
http://127.0.0.1:8090/a
לא קורה כלום

[nigun]

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

func someThread(w http.ResponseWriter, r *http.Request) {
        fmt.Fprintf(w, "")
        go func() {
                c := addListener()
                s := <-c
                fmt.Println(s)
//קודם כל אני סוגר את הערוץ (לא בטוח שזה חכם)
                close(c)
// אני מריץ ללואה שמסירה את הערוץ המת מהמערך
                for i := 0; i < len(channels); i++ {
                        chanvar := channels[i]
                        if chanvar == c {
                                channels = append(channels[:i], channels[i+1:]...)
                                i-- // Important: decrease index
                                break
                        }

                }
        }()
}

השאלה היא מה יקרה אם יגיע קריאה חדשה לערוץ הראשי תוך כדי שאני סוגר את הערוץ לפני שאני מסיר אותו מהמערך
לכאורה אני יקבל שגיאה

panic: send on closed channel

כי אני מנסה לשלוח מידע לערוץ סגור
הפרון הוא אולי לנעול את המאזין הראשי כשאני מסיר איבר מהמערך
השאלה היא איך אני מדבג את זה?
לבניים אני מוותר על סגירת הערוץ, ורק מוחק אותו מהמערך
השאלה היא מה ההשלכות?

[יוסף בן שמעון]

@nigun אמר בהאזנה לערוץ בתוך כמה פונקציות במקביל:

לבניים אני מוותר על סגירת הערוץ, ורק מוחק אותו מהמערך
השאלה היא מה ההשלכות?

לפי מה שכתוב פה
https://stackoverflow.com/questions/8593645/is-it-ok-to-leave-a-channel-open
אין צורך לסגור את הערוץ, מספיק לנקות את המצביע אליו ואיסוף הזבל ידאג לשאר

[יוסף בן שמעון]

@nigun לכאורה אתה צודק, אם זה המצב אז הקוד כבר מרגיש לי מידי מסטיקי, אולי תשקול לעבוד עם קולבקים פרימיטייבים? או לחילופין לממש את הרעיון של סגירת ערוץ שתשלח הודעה לכולם, ואת הערך שאתה רוצה להעביר תעביר עם איזה משתנה גלובלי?

[nigun]

@יוסף-בן-שמעון
איך אני יאזין למשתנה בין הערוצים? עם האזנה אינסופית עם if ובמקרה שהמשתנה שונה מהקריאה הקודמת לעשות פעולה כל שהיא?

[יוסף בן שמעון]

@nigun אני מדבר על משהו כזה

package main

import (
	"fmt"
	"time"
)

var data string

func main() {
	c := make(chan string)

	go func(c chan string) {
		time.Sleep(2 * time.Second)
		data = "foo"
		close(c)

	}(c)

	go func() {
		<-c
		fmt.Println(data)

	}()
	go func() {
		<-c
		fmt.Println(data)

	}()

	time.Sleep(4 * time.Second)
}

[nigun]

@יוסף-בן-שמעון
אני צריך שליחה רב פעמית
זאת אומרת שאני יוסיף ויוריד מאזינים (שיוכלו להאזין פעם אחת או יותר)
ולשלוח כל הזמן (עם שרת HTTP ) מידע שיעבור לכל המאזינים
את האמת אני צריך לשלוח את המידע לרוב רק למאזין אחד, (כי כל מאזין שייך למשתמש אחר)
ואני שולח את הבקשה בצורה כזאת USER:DATA ולמאזין אני בונה לולאה שמאזינה כל הזמן לערוץ
ואם היוזר שווה ליוזר שלו הוא עושה את הפעולה המבוקשת ואם לא הוא חוזר להאזין,
האם אני עובד נכון או לא?

[יוסף בן שמעון]

@nigun אמר בהאזנה לערוץ בתוך כמה פונקציות במקביל:

האם אני עובד נכון או לא?

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

[nigun]

@יוסף-בן-שמעון
אני לרוב מסתבך כשכותבים מידי הרבה פרטים, אבל אם זה יעזור אז מצויין.
אני בונה מרכזיה באסטריסק שבו אני רוצה לשלוט על חלק מהמאזינים
דהיינו לדחוף להם באמצע האזנה פקודות שונות דרך API משרת HTTP
(אמנם יש לאסטריסק API לשליחת פקודות כאלו ,אבל אני לא הצלחתי להשתמש בו למה שאני צריך)
לכן חשבתי ללכת על הכיוון של ערוצי GO שפשוט כל שיחה (שאני רוצה לשלוט עליה) תאזין לערוץ
ובAPI החיצוני אני ישלח את שם המאזין שבו אני רוצה לשלוט , ואת הפקודה שאני רוצה שהוא יבצע
כשמתקבל בקשה כל המאזינים יבצעו בדיקה האם הבקשה מופנית אליהם, ואם כן יבצעו אותה
לקינוח אני יכול להחזיר באותה צורה הודעה דרך ערוץ אחר לשרת HTTP שהפעולה התבצעה +מידע על התוצאה.

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