使用互斥体在 goroutine 之间修改的切片未显示正确的同步

Golang不知道大家是否熟悉？今天我将给大家介绍《使用互斥体在 goroutine 之间修改的切片未显示正确的同步》，这篇文章主要会讲到等等知识点，如果你在看完本篇文章后，有更好的建议或者发现哪里有问题，希望大家都能积极评论指出，谢谢！希望我们能一起加油进步！

问题内容

我是新手，但之前接触过并发。我在多个 goroutine 之间共享切片时遇到问题，但所有 goroutine 之间不包含相同的数据。当我修改切片时，我也使用互斥体来锁定结构，但它似乎没有帮助。我已附上我的代码，想知道我做错了什么，感谢您的帮助！

type state struct {
    waiting int32
    processing int32
    completed int32
}

type scheduler struct {
    sync.mutex
    items chan interface{}
    backpressure []interface{}
    capacity int
    canceler context.cancelfunc
    state state
}

func newscheduler(capacity int, handler func(interface {}) (interface{}, error)) scheduler {
    ctx, cancel := context.withcancel(context.background())

    state := state{}

    atomic.storeint32(&state.waiting, 0)
    atomic.storeint32(&state.processing, 0)
    atomic.storeint32(&state.completed, 0)

    scheduler := scheduler{
        items: make(chan interface{}, capacity),
        backpressure: make([]interface{}, 0),
        capacity: capacity,
        canceler: cancel,
        state: state,
    }

    scheduler.initializeworkers(ctx, handler)

    return scheduler
}

func (s *scheduler) initializeworkers(ctx context.context, handler func(interface {}) (interface{}, error)) {
    for i := 0; i < 5; i++ {
        go s.newworker(ctx, handler)
    }
}

func (s *scheduler) newworker(ctx context.context, handler func(interface {}) (interface{}, error)) {
    backoff := 0

    for {
        select {
        case <-ctx.done():
            return
        case job := <- s.items:
            atomic.addint32(&s.state.waiting, -1)
            atomic.addint32(&s.state.processing, 1)
            job, _ = handler(job)
            backoff = 0
            atomic.addint32(&s.state.processing, -1)
            atomic.addint32(&s.state.completed, 1)
        default:
            backoff += 1
            s.checkbackpressure()
            time.sleep(time.duration(backoff * 10) * time.millisecond)
        }
    }
}

func (s *scheduler) additem(item interface{}) {
    atomic.addint32(&s.state.waiting, 1)

    if len(s.items) < s.capacity {
        select {
        case s.items <- item:
            return
        }
    }

    s.lock()
    defer s.unlock()

    s.backpressure = append(s.backpressure, item)

    fmt.printf("new backpressure len %v \n", len(s.backpressure))

    return
}

func (s *scheduler) process() {
    var wg sync.waitgroup

    wg.add(1)


    go func() {
        defer wg.done()

        for {
            if atomic.loadint32(&s.state.waiting) == 0 && atomic.loadint32(&s.state.processing) == 0 {
                return
            }
            runtime.gosched()
        }
    }()

    wg.wait()
}

func (s *scheduler) checkbackpressure() {
    s.lock()
    defer s.unlock()

    if len(s.backpressure) == 0 || s.capacity <= len(s.items) {
        fmt.printf("backpressure = %d  :: len = %d cap = %d \n", len(s.backpressure), len(s.items), s.capacity)
        return
    }

    fmt.printf("releasing backpressure \n")

    job, tmp := s.backpressure[0], s.backpressure[1:]

    s.backpressure = tmp

    s.items <- job
    return
}

func (s *scheduler) stop() {
    s.canceler()
}

这是我用来测试功能的代码：

type job struct {
    value int
}

func testschedulerexceedingcapacity(t *testing.t) {


    handler := func (ptr interface{}) (interface{}, error) {
        job, ok := (ptr).(*job)

        if ok != true {
            return nil, errors.new("failed to convert job")
        }

        // simulate work
        time.sleep(50 * time.millisecond)

        return job, nil
    }

    scheduler := newscheduler(5, handler)

    for i := 0; i < 25; i++ {
        scheduler.additem(&(job { value: i }))
    }

    fmt.printf("processing\n")
    scheduler.process()
    fmt.printf("finished\n")
}

当我更新保持背压的切片时，似乎表明通过打印 new backpressure len 1 for 1-16 已正确更新。

但是，当我检查worker的背压时，表明背压片是空的。 背压 = 0 :: len = 0 上限 = 5。

此外，“释放背压”也永远不会打印到标准输出。

这是一些额外的输出...

=== RUN   TestSchedulerExceedingCapacity
new backpressure len 1 
new backpressure len 2 
new backpressure len 3 
new backpressure len 4 
new backpressure len 5 
new backpressure len 6 
new backpressure len 7 
new backpressure len 8 
backpressure = 0  :: len = 0 cap = 5 
new backpressure len 9 
new backpressure len 10 
new backpressure len 11 
new backpressure len 12 
new backpressure len 13 
new backpressure len 14 
new backpressure len 15 
new backpressure len 16 
PROCESSING
backpressure = 0  :: len = 0 cap = 5 
backpressure = 0  :: len = 0 cap = 5 
backpressure = 0  :: len = 0 cap = 5 
...

如果我不终止测试，它会无限期地打印 backpressure = 0 :: len = 0 cap = 5

我假设我没有正确同步更改，我真的很感激任何见解，谢谢！

解决方案

好吧，当我发布问题后，我当然能够解决这个问题......

我在某处看到建议使用启用 data race detector 的 -race 选项运行测试。我立即收到错误，这有助于使问题更容易调试。

事实证明，问题与返回newscheduler的值有关，而不是与新调度程序的指针有关。我将该函数更改为以下代码，解决了该问题。

func NewScheduler(capacity int, handler func(interface {}) (interface{}, error)) *Scheduler {
    ctx, cancel := context.WithCancel(context.Background())

    state := State{}

    atomic.StoreInt32(&state.waiting, 0)
    atomic.StoreInt32(&state.processing, 0)
    atomic.StoreInt32(&state.completed, 0)
    atomic.StoreInt32(&state.errors, 0)

    scheduler := Scheduler{
        items: make(chan interface{}, capacity),
        backPressure: make([]interface{}, 0),
        capacity: capacity,
        canceler: cancel,
        state: state,
    }

    scheduler.initializeWorkers(ctx, handler)

    return &scheduler
}

本篇关于《使用互斥体在 goroutine 之间修改的切片未显示正确的同步》的介绍就到此结束啦，但是学无止境，想要了解学习更多关于Golang的相关知识，请关注编程网公众号！