Go言語のsync.Condの挙動が分かりません。

Question

該当のソースコード

GitHubのソースコード

発生している問題・エラーメッセージ

期待する結果は以下の通りなのですが、

{"success":true,"id":1}
{"success":true,"id":2}
{"success":true,"id":3}
{"success":false,"id":4}
{"success":true,"id":5}
{"success":false,"id":6}
{"id":7,"feed":[{"body":"Another post to add","timestamp":43242421}]}

以下のコマンドで2つのサブスレッドを発生させると（一つ目のコマンド引数がスレッドの数です。）、デッドロックを起こしてしまいます。

$ twitter kazukiegusa$ go run twitter.go 2 4 < tasks.json 
{"success":true,"id":1}
{"success":true,"id":2}
{"success":true,"id":3}
{"success":false,"id":4}
{"success":false,"id":6}
{"id":7,"feed":[{"body":"Another post to add","timestamp":43242421}]}
fatal error: all goroutines are asleep - deadlock!

goroutine 1 [semacquire]:
sync.runtime_Semacquire(0xc000016088)
        /usr/local/go/src/runtime/sema.go:56 +0x42
sync.(*WaitGroup).Wait(0xc000016080)
        /usr/local/go/src/sync/waitgroup.go:130 +0x64
main.main()
        /Users/kazukiegusa/mpcs/Parallel Programming/egusa/proj1/twitter/twitter.go:66 +0x422
exit status 2

実現したいこと＆わからないこと

質問の箇所は、twitter.go内にあります。まず、main関数の中で、コンディション変数を定義しています。

・・・中略・・・

var wg sync.WaitGroup

cond := sync.NewCond(new(sync.Mutex)) // Conditional variable

for i := 0; i < numGoroutines; i++ {
    wg.Add(1)
    go consumer(q, f, blockSize, cond, &wg)
}

producer(q, cond)

// 全てのサブスレッドがconsumer関数の実行を終えるまで待つ
wg.Wait()

producer関数の中では、エンキュー（Enqueue）するごとに、sync.CondのSignal関数を使って、consumer関数内で待機しているスレッドを一つ呼び起こしています。

func producer(q *queue.Queue, cond *sync.Cond) {

・・・中略・・・

    for {

        ・・・中略（ここにエンキュー（Enqueue）を実行するコードが入ります。）・・・

		// 待機中のサブスレッドを一つ呼び起こす
		cond.Signal()
	}
}

consumer関数内では、初期化するときにsync.Condに紐付けたMutexを使ってロックしています。

func consumer(q *queue.Queue, f feed.Feed, blockSize int64, cond *sync.Cond, wg *sync.WaitGroup) {

	cond.L.Lock()

	// キューが空なら待機する
	cond.Wait()

    ・・・中略（ここにデキュー（Dequeue）を実行するコードが入ります。）・・・

	cond.L.Unlock()

	wg.Done()
}

試したこと

いろいろ試しましたが、sync.Condの挙動がきちんと把握できていないのかなと思っています。

よろしくお願い致します。

Comments

[nobonobo]

タイトル通りの質問内容にまとめて欲しい・・・。
sync.Condにどのような挙動を期待していて試してみたことに絞って質問してもらえれば回答がつきやすいと思います。

[退会済みユーザー]

ご指摘ありがとうございます！質問を修正してみました。

コードも多少変えています。どうやらデッドロックが起きてしまっているみたいです。

Answer 1

sync.Condの使い方は https://golang.org/pkg/sync/#Cond.Wait

その他の問題はいろいろ指摘が多岐にわたってしまうのでdiffだけ提供します。
（というか質問の際に抜粋等してもらえると回答しやすいです）

diff --git a/queue/queue.go b/queue/queue.go
index d8b1c50..2af95b7 100644
--- a/queue/queue.go
+++ b/queue/queue.go
@@ -27,6 +27,11 @@ func NewQueue() (q *Queue) {
 	return
 }
 
+// IsEmpty ...
+func (q *Queue) IsEmpty() bool {
+	return q.head == q.tail
+}
+
 // LoadPointer atomically loads *addr.
 // https://golang.org/pkg/sync/atomic/#LoadPointer
 func load(p *unsafe.Pointer) (n *node) {
diff --git a/twitter/twitter.go b/twitter/twitter.go
index 242a0f6..c9c81aa 100644
--- a/twitter/twitter.go
+++ b/twitter/twitter.go
@@ -47,7 +47,7 @@ func main() {
 	if len(args) == 0 {
 		blockSize := int64(math.MaxInt64)
 		producer(q, nil)
-		consumer(q, f, blockSize, nil, &wg)
+		consumer(q, f, blockSize, nil)
 
 	} else {
 		numGoroutines, _ := strconv.Atoi(args[0])         // Number of goroutines to spawn
@@ -57,7 +57,10 @@ func main() {
 
 		for i := 0; i < numGoroutines; i++ {
 			wg.Add(1)
-			go consumer(q, f, blockSize, cond, &wg)
+			go func() {
+				defer wg.Done()
+				consumer(q, f, blockSize, cond)
+			}()
 		}
 
 		producer(q, cond)
@@ -93,13 +96,14 @@ func producer(q *queue.Queue, cond *sync.Cond) {
 }
 
 // consumer grabs tasks from the queue and starts to execute them.
-func consumer(q *queue.Queue, f feed.Feed, blockSize int64, cond *sync.Cond, wg *sync.WaitGroup) {
+func consumer(q *queue.Queue, f feed.Feed, blockSize int64, cond *sync.Cond) {
 
 	if cond != nil {
 		cond.L.Lock()
-
-		// Go to sleep if the queue is empty
-		cond.Wait()
+		for q.IsEmpty() {
+			cond.Wait()
+		}
+		cond.L.Unlock()
 	}
 
 	// Dequeue the task
@@ -132,9 +136,8 @@ func consumer(q *queue.Queue, f feed.Feed, blockSize int64, cond *sync.Cond, wg
 	}
 
 	if cond != nil {
-		cond.L.Unlock()
+		//cond.L.Unlock()
 	}
-	wg.Done()
 }
 
 // response execute the task.

Comments

[nobonobo]

質問のリポジトリを消されると僕の回答も無意味になってしまいますね。今後は質問されるときはここに書かれた情報で完結するような質問にまとめていただけると助かります！

[退会済みユーザー]

すいません、かしこまりました！次回から消さないようにします。