2

これが私のコードです：

package main

import (
    "sync/atomic"
    "unsafe"
    "sync"
    "fmt"
    "time"
)

const (
    MAX_DATA_SIZE = 100
)

// lock free queue
type Queue struct {
    head unsafe.Pointer
    tail unsafe.Pointer
}
// one node in queue
type Node struct {
    val interface{}
    next unsafe.Pointer
}
// queue functions
func (self *Queue) enQueue(val interface{}) {
    newValue := unsafe.Pointer(&Node{val: val, next: nil})
    var tail,next unsafe.Pointer
    for {
        tail = self.tail
        next = ((*Node)(tail)).next
        if next != nil {
            atomic.CompareAndSwapPointer(&(self.tail), tail, next)
        }else if atomic.CompareAndSwapPointer(&((*Node)(tail).next), nil, newValue){
            break
        }
    }
}

func (self *Queue) deQueue() (val interface{}, success bool){
    var head,tail,next unsafe.Pointer
    for {
        head = self.head
        tail = self.tail
        next = ((*Node)(head)).next
        if head == tail {
            if next == nil {
                return nil, false
            }else {
                atomic.CompareAndSwapPointer(&(self.tail), tail, next)
            }
        }else {
            val = ((*Node)(next)).val
            if atomic.CompareAndSwapPointer(&(self.head), head, next) {
                return val, true
            }
        }
    }
    return
}

func main() {
    var wg sync.WaitGroup
    wg.Add(20)
    queue := new(Queue)
    queue.head = unsafe.Pointer(new(Node))
    queue.tail = queue.head

    for i := 0; i < 10; i++ {
        go func() {
            defer wg.Done()
            for j := 0; j < MAX_DATA_SIZE; j++ {
                t := time.Now()
                queue.enQueue(t)
                fmt.Println("enq = ", t)
            }
        }()
    }

    for i := 0; i < 10; i++ {
        go func() {
            ok := false
            var val interface{}
            defer wg.Done()
            for j := 0; j < MAX_DATA_SIZE; j++ {
                val,ok = queue.deQueue()
                for !ok {
                    val,ok = queue.deQueue()
                }
                fmt.Println("deq = ",val)
            }
        }()
    }

    wg.Wait()
}

問題は、コードが正常に実行されることもありますが、失敗して応答がないままスタックすることもあります。

コードに問題はありますか？

4

2 に答える 2

5

このコードには多くのアクティブな待機があり、Nickの素晴らしいコードと同じようにチャネルをクリーンに使用することを強くお勧めします。

ただし、「なぜスタックしているのか」という元の質問に対する私の答えは次のとおりです。：各ゴルーチンが他のゴルーチンを実行させるためにいつ降伏するかについての保証はありません。おそらく、無限ループ内にあるときに降伏することはありません。

これを修正するには、無限ループの可能性がある各forループ内でruntime.Gosched（）を使用します。

Goschedはプロセッサを生成し、他のgoroutineを実行できるようにします。現在のゴルーチンを一時停止しないため、実行が自動的に再開されます。

この拡張されたコードは、元のコードとほぼ同じ速度で実行されますが、ハングすることはありません。

package main

import (
    "fmt"
    "runtime"
    "sync"
    "sync/atomic"
    "time"
    "unsafe"
)

const (
    MAX_DATA_SIZE = 100
)

// lock free queue
type Queue struct {
    head unsafe.Pointer
    tail unsafe.Pointer
}

// one node in queue
type Node struct {
    val  interface{}
    next unsafe.Pointer
}

// queue functions
func (self *Queue) enQueue(val interface{}) {
    newValue := unsafe.Pointer(&Node{val: val, next: nil})
    var tail, next unsafe.Pointer
    for {
        tail = self.tail
        next = ((*Node)(tail)).next
        if next != nil {
            atomic.CompareAndSwapPointer(&(self.tail), tail, next)
        } else if atomic.CompareAndSwapPointer(&((*Node)(tail).next), nil, newValue) {
            break
        }
        runtime.Gosched()
    }
}

func (self *Queue) deQueue() (val interface{}, success bool) {
    var head, tail, next unsafe.Pointer
    for {
        head = self.head
        tail = self.tail
        next = ((*Node)(head)).next
        if head == tail {
            if next == nil {
                return nil, false
            } else {
                atomic.CompareAndSwapPointer(&(self.tail), tail, next)
            }
        } else {
            val = ((*Node)(next)).val
            if atomic.CompareAndSwapPointer(&(self.head), head, next) {
                return val, true
            }
        }
        runtime.Gosched()
    }
    return
}

func main() {
    var wg sync.WaitGroup
    wg.Add(20)
    queue := new(Queue)
    queue.head = unsafe.Pointer(new(Node))
    queue.tail = queue.head

    for i := 0; i < 10; i++ {
        go func() {
            defer wg.Done()
            for j := 0; j < MAX_DATA_SIZE; j++ {
                t := time.Now()
                queue.enQueue(t)
                fmt.Println("enq = ", t)
            }
        }()
    }

    for i := 0; i < 10; i++ {
        go func() {
            ok := false
            var val interface{}
            defer wg.Done()
            for j := 0; j < MAX_DATA_SIZE; j++ {
                val, ok = queue.deQueue()
                for !ok {
                    val, ok = queue.deQueue()
                    runtime.Gosched()
                }
                fmt.Println("deq = ", val)
            }
        }()
    }

    wg.Wait()
}
于 2013-01-21T13:52:25.053 に答える
3

これは、@ mkbが提案したようにチャネルで書き直された上記のとおりです（無限のキューサイズを除く）。

ロックアップしません。

Goチームが信頼性が高く、高性能で使いやすいようにするために多大な労力を費やしてきたため、本当に正当な理由がない限り、チャネルを使用することをお勧めします。

package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

const (
    MAX_DATA_SIZE = 100
)

func main() {
    runtime.GOMAXPROCS(4)
    var wg sync.WaitGroup
    wg.Add(20)
    queue := make(chan time.Time, 10)

    for i := 0; i < 10; i++ {
        go func() {
            defer wg.Done()
            for j := 0; j < MAX_DATA_SIZE; j++ {
                t := time.Now()
                queue <- t
                fmt.Println("enq = ", t)
            }
        }()
    }

    for i := 0; i < 10; i++ {
        go func() {
            defer wg.Done()
            for j := 0; j < MAX_DATA_SIZE; j++ {
                val := <-queue
                fmt.Println("deq = ", val)
            }
        }()
    }

    wg.Wait()
}
于 2012-09-06T18:23:11.247 に答える