गो में निर्माता/उपभोक्ता के लिए सबसे अच्छा मुहावरे क्या है?

Question

मैं जो करना चाहता हूं उसके पास निर्माता गोरौटाइन का एक सेट है (जिनमें से कुछ पूरा हो सकता है या नहीं) और एक उपभोक्ता दिनचर्या है। मुद्दा उस चेतावनी के साथ है जो कोष्ठक में है - हम कुल संख्या को नहीं जानते हैं जो एक जवाब वापस कर देगा।

तो मुझे क्या करना चाहते हैं यह है:

package main 

import (
    "fmt" 
    "math/rand" 
) 

func producer(c chan int) { 
    // May or may not produce. 
    success := rand.Float32() > 0.5 
    if success { 
    c <- rand.Int() 
    } 
} 

func main() { 
    c := make(chan int, 10) 
    for i := 0; i < 10; i++ { 
    go producer(c, signal) 
    } 

    // If we include a close, then that's WRONG. Chan will be closed 
    // but a producer will try to write to it. Runtime error. 
    close(c) 

    // If we don't close, then that's WRONG. All goroutines will 
    // deadlock, since the range keyword will look for a close. 
    for num := range c { 
    fmt.Printf("Producer produced: %d\n", num) 
    } 
    fmt.Println("All done.") 
} 

तो मुद्दा अगर मैं इसे बंद यह गलत है, अगर मैं बंद नहीं करते हैं, है - यह अभी भी गलत है (कोड में टिप्पणियां देखें)।

अब, समाधान एक आउट-ऑफ-बैंड संकेत चैनल होगा, कि सभी उत्पादकों लिखने के लिए:

package main 

import (
    "fmt" 
    "math/rand" 
) 

func producer(c chan int, signal chan bool) { 
    success := rand.Float32() > 0.5 
    if success { 
    c <- rand.Int() 
    } 
    signal <- true 
} 

func main() { 
    c := make(chan int, 10) 
    signal := make(chan bool, 10) 
    for i := 0; i < 10; i++ { 
    go producer(c, signal) 
    } 

    // This is basically a 'join'. 
    num_done := 0 
    for num_done < 10 { 
    <- signal 
    num_done++ 
    } 
    close(c) 

    for num := range c { 
    fmt.Printf("Producer produced: %d\n", num) 
    } 
    fmt.Println("All done.") 
} 

और वह पूरी तरह से जो मैं चाहता करता है! लेकिन मेरे लिए यह एक मुंह की तरह लगता है। मेरा सवाल है: क्या कोई मुहावरे/चाल है जो मुझे कुछ आसान तरीके से करने देती है?

मैं यहाँ एक नज़र था: http://golang.org/doc/codewalk/sharemem/ और यह complete चान (main के शुरू में initialised) एक श्रेणी में प्रयोग किया जाता है लेकिन कभी नहीं बंद कर दिया की तरह लगता है। मुझे समझ में नहीं आता कि कैसे।

यदि किसी के पास कोई अंतर्दृष्टि है, तो मैं इसकी सराहना करता हूं। चीयर्स!

---

संपादित करें: fls0815 जवाब है, और भी कितने करीब-कम चैनल रेंज काम करता है के सवाल का जवाब दे गया है।

काम करने के लिए modifed ऊपर मेरे कोड (पहले fls0815 कृपया आपूर्ति कोड किया):

package main 

import (
    "fmt" 
    "math/rand" 
    "sync" 
) 

var wg_prod sync.WaitGroup 
var wg_cons sync.WaitGroup 

func producer(c chan int) { 
    success := rand.Float32() > 0.5 
    if success { 
    c <- rand.Int() 
    } 
    wg_prod.Done() 
} 

func main() { 
    c := make(chan int, 10) 
    wg_prod.Add(10) 
    for i := 0; i < 10; i++ { 
    go producer(c) 
    } 

    wg_cons.Add(1) 
    go func() { 
    for num := range c { 
     fmt.Printf("Producer produced: %d\n", num) 
    } 
    wg_cons.Done() 
    }() 

    wg_prod.Wait() 
    close(c) 
    wg_cons.Wait() 
    fmt.Println("All done.") 
} 

Answer 1

केवल उत्पादकों चैनलों बंद हो जाना चाहिए। एक बार आपके निर्माता शुरू होने के बाद परिणामस्वरूप चैनल पर उपभोक्ता (ओं) को आमंत्रित करके आप अपना लक्ष्य प्राप्त कर सकते हैं (range)। आपके मुख्य थ्रेड में आप प्रतीक्षा करें (sync.WaitGroup देखें) जब तक आपके उपभोक्ता/उत्पादक अपना काम पूरा नहीं कर लेते। उत्पादकों के समाप्त होने के बाद आप परिणामस्वरूप चैनल बंद कर देंगे जो आपके उपभोक्ताओं को बाहर निकलने के लिए मजबूर करेगा (range चैनल बंद होने पर बाहर निकलेंगे और कोई बफर किए गए आइटम को छोड़ दिया जाएगा)।

उदाहरण कोड:

package main 

import (
    "log" 
    "sync" 
    "time" 
    "math/rand" 
    "runtime" 
) 

func consumer() { 
    defer consumer_wg.Done() 

    for item := range resultingChannel { 
     log.Println("Consumed:", item) 
    } 
} 

func producer() { 
    defer producer_wg.Done() 

    success := rand.Float32() > 0.5 
    if success { 
     resultingChannel <- rand.Int() 
    } 
} 

var resultingChannel = make(chan int) 
var producer_wg sync.WaitGroup 
var consumer_wg sync.WaitGroup 

func main() { 
    rand.Seed(time.Now().Unix()) 

    for c := 0; c < runtime.NumCPU(); c++ { 
     producer_wg.Add(1) 
     go producer() 
    } 

    for c := 0; c < runtime.NumCPU(); c++ { 
     consumer_wg.Add(1) 
     go consumer() 
    } 

    producer_wg.Wait() 

    close(resultingChannel) 

    consumer_wg.Wait() 
} 

क्योंकि हम एक से अधिक उत्पादक है कारण मैं मुख्य समारोह में close -statement डाल है। उपर्युक्त उदाहरण में एक निर्माता में चैनल को बंद करने से आप जिस समस्या से पहले ही भाग चुके हैं (बंद चैनलों पर लिखना; कारण यह है कि एक निर्माता छोड़ सकता है जो अभी भी डेटा उत्पन्न करता है)। जब कोई उत्पादक नहीं छोड़ा जाता है तो चैनल केवल तभी बंद होना चाहिए (इसलिए निर्माता द्वारा केवल चैनल को बंद करने पर मेरा सुझाव)। इस तरह चैनलों में गो का निर्माण किया जाता है। Here आपको चैनल बंद करने पर कुछ और जानकारी मिल जाएगी।

---

sharemem उदाहरण से संबंधित: AFAICS इस उदाहरण संसाधन बार-बार फिर से कतार से अंतहीन रन (से लंबित -> पूरा -> लंबित -> पूरा ... और इसी तरह)। मुख्य-func के अंत में यह पुनरावृत्ति है। यह पूर्ण संसाधन प्राप्त करता है और लंबित करने के लिए संसाधन। नींद() का उपयोग करके उन्हें फिर से पंक्तिबद्ध करता है। जब कोई पूरा संसाधन नहीं होता है तो यह पूरा होने के लिए नए संसाधनों की प्रतीक्षा करता है और ब्लॉक करता है। इसलिए चैनल बंद करने की कोई आवश्यकता नहीं है क्योंकि वे हर समय उपयोग में हैं।

Answer 2

इन समस्याओं को हल करने के लिए हमेशा कई तरीके हैं। यहां सरल सिंक्रोनस चैनलों का उपयोग करके एक समाधान है जो गो में मौलिक हैं। कोई buffered चैनल, कोई बंद चैनल, कोई प्रतीक्षा समूह नहीं।

यह वास्तव में आपके "मुंह से" समाधान से बहुत दूर नहीं है, और - निराश होने के लिए खेद है - इतना छोटा नहीं। यह उपभोक्ता को अपने स्वयं के गोरौटाइन में डाल देता है, ताकि उपभोक्ता संख्याओं का उपभोग कर सके क्योंकि निर्माता उन्हें उत्पादित करता है। यह भी भेद बनाता है कि उत्पादन "कोशिश" सफलता या विफलता में समाप्त हो सकता है। अगर उत्पादन विफल रहता है, तो प्रयास तुरंत किया जाता है। यदि यह सफल होता है, तब तक प्रयास नहीं किया जाता है जब तक कि संख्या का उपभोग न हो जाए।

package main 

import (
    "fmt" 
    "math/rand" 
) 

func producer(c chan int, fail chan bool) { 
    if success := rand.Float32() > 0.5; success { 
     c <- rand.Int() 
    } else { 
     fail <- true 
    } 
} 

func consumer(c chan int, success chan bool) { 
    for { 
     num := <-c 
     fmt.Printf("Producer produced: %d\n", num) 
     success <- true 
    } 
} 

func main() { 
    const nTries = 10 
    c := make(chan int) 
    done := make(chan bool) 
    for i := 0; i < nTries; i++ { 
     go producer(c, done) 
    } 
    go consumer(c, done) 

    for i := 0; i < nTries; i++ { 
     <-done 
    } 
    fmt.Println("All done.") 
} 

Answer 3

मैं इसे जोड़ रहा हूं क्योंकि मौजूदा उत्तर कुछ चीजों को स्पष्ट नहीं करते हैं। सबसे पहले, कोडवॉक उदाहरण में रेंज लूप केवल एक अनंत घटना लूप है, फिर भी वही यूआरएल सूची को दोबारा जांचने और अपडेट करने के लिए।

अगला, एक चैनल, सब कुछ, पहले से ही गो में बेवकूफ उपभोक्ता-उत्पादक कतार है। चैनल का समर्थन करने वाले एसिंक बफर का आकार यह निर्धारित करता है कि बैकप्रेसर प्राप्त करने से पहले उत्पादक कितने उत्पादन कर सकते हैं। लॉक-स्टेप उत्पादक उपभोक्ता को आगे बढ़ने या पीछे आने के बिना नीचे एन = 0 सेट करें। जैसा कि है, एन = 10 निर्माता को अवरुद्ध करने से पहले 10 उत्पादों तक उत्पादन करने देगा।

आखिरकार, गो में अनुक्रमिक प्रक्रियाओं को संचारित करने के लिए कुछ अच्छे मुहावरे हैं (जैसे फ़ंक्शन जो आपके लिए दिनचर्या शुरू करते हैं, और नियंत्रण आदेशों को संवाद करने और स्वीकार करने के लिए/चयन पैटर्न का उपयोग करते हैं)। मैं प्रतीक्षा समूह को बेकार के रूप में सोचता हूं, और इसके बजाय बेवकूफ उदाहरण देखना चाहता हूं।

package main 

import (
    "fmt" 
    "time" 
) 

type control int 
const (
    sleep control = iota 
    die // receiver will close the control chan in response to die, to ack. 
) 

func (cmd control) String() string { 
    switch cmd { 
    case sleep: return "sleep" 
    case die: return "die" 
    } 
    return fmt.Sprintf("%d",cmd) 
} 

func ProduceTo(writechan chan<- int, ctrl chan control, done chan bool) { 
    var product int 
    go func() { 
     for { 
      select { 
     case writechan <- product: 
      fmt.Printf("Producer produced %v\n", product) 
      product++ 
     case cmd:= <- ctrl: 
      fmt.Printf("Producer got control cmd: %v\n", cmd) 
      switch cmd { 
      case sleep: 
       fmt.Printf("Producer sleeping 2 sec.\n") 
       time.Sleep(2000 * time.Millisecond) 
      case die: 
       fmt.Printf("Producer dies.\n") 
       close(done) 
       return 
      } 
      } 
     } 
    }() 
} 

func ConsumeFrom(readchan <-chan int, ctrl chan control, done chan bool) { 
    go func() { 
     var product int 
     for { 
      select { 
      case product = <-readchan: 
       fmt.Printf("Consumer consumed %v\n", product) 
      case cmd:= <- ctrl: 
       fmt.Printf("Consumer got control cmd: %v\n", cmd) 
       switch cmd { 
       case sleep: 
        fmt.Printf("Consumer sleeping 2 sec.\n") 
        time.Sleep(2000 * time.Millisecond) 
       case die: 
        fmt.Printf("Consumer dies.\n") 
        close(done) 
        return 
       } 

      } 
     } 
    }() 
} 

func main() { 

    N := 10 
    q := make(chan int, N) 

    prodCtrl := make(chan control) 
    consCtrl := make(chan control) 

    prodDone := make(chan bool) 
    consDone := make(chan bool) 


    ProduceTo(q, prodCtrl, prodDone) 
    ConsumeFrom(q, consCtrl, consDone) 

    // wait for a moment, to let them produce and consume 
    timer := time.NewTimer(10 * time.Millisecond) 
    <-timer.C 

    // tell producer to pause 
    fmt.Printf("telling producer to pause\n") 
    prodCtrl <- sleep 

    // wait for a second 
    timer = time.NewTimer(1 * time.Second) 
    <-timer.C 

    // tell consumer to pause 
    fmt.Printf("telling consumer to pause\n") 
    consCtrl <- sleep 


    // tell them both to finish 
    prodCtrl <- die 
    consCtrl <- die 

    // wait for that to actually happen 
    <-prodDone 
    <-consDone 
} 

Answer 4

आप इंतजार समूहों के बिना सरल unbuffered चैनलों का उपयोग करता है, तो आप एक fanIn समारोह के साथ जनरेटर पैटर्न का उपयोग कर सकते हैं।

जेनरेटर पैटर्न में, प्रत्येक निर्माता एक चैनल देता है और इसे बंद करने के लिए ज़िम्मेदार है। एक fanIn फ़ंक्शन फिर इन चैनलों पर फिर से चलाता है और आगे उन मानों को वापस लौटाता है जो इसे लौटाते हैं।

पाठ्यक्रम की समस्या यह है कि प्रत्येक चैनल बंद होने पर fanIn फ़ंक्शन चैनल प्रकार (int) के शून्य मान को आगे बढ़ाता है।

आप अपने चैनल प्रकार के शून्य मान को सेंटीनेल मान के रूप में उपयोग करके और केवल प्रशंसक चैनल के परिणामों का उपयोग करके इसके आसपास काम कर सकते हैं यदि वे शून्य मान नहीं हैं।

यहाँ एक उदाहरण है:

package main 

import (
    "fmt" 
    "math/rand" 
) 

const offset = 1 

func producer() chan int { 
    cout := make(chan int) 
    go func() { 
     defer close(cout) 
     // May or may not produce. 
     success := rand.Float32() > 0.5 
     if success { 
      cout <- rand.Int() + offset 
     } 
    }() 
    return cout 
} 

func fanIn(cin []chan int) chan int { 
    cout := make(chan int) 
    go func() { 
     defer close(cout) 
     for _, c := range cin { 
      cout <- <-c 
     } 
    }() 
    return cout 
} 

func main() { 
    chans := make([]chan int, 0) 
    for i := 0; i < 10; i++ { 
     chans = append(chans, producer()) 
    } 

    for num := range fanIn(chans) { 
     if num > offset { 
      fmt.Printf("Producer produced: %d\n", num) 
     } 
    } 
    fmt.Println("All done.") 
}