Как ждать потоков с низкой задержкой в ​​go?

Question

Я пытался создать простую оболочку цикла событий в Go. Но я был озадачен, как я должен был отслеживать операции в текущем потоке? Я хотел, чтобы CurrentTick запускал функцию, и даже если вызывающая функция закрывается, не запускайте следующий тик, пока все функции, запущенные CurrentTick, не завершат работу. Я подумал, что мог бы использовать мьютекс для отслеживания количества потоков, но я понял, что если я буду продолжать проверять это снова и снова, это приведет к снижению производительности процессора. Если бы я использовал время. Сон был бы скрыт. Как бы вы решили проблему?

package eventloop

import (
    "reflect"
)

type eventLoop *struct{
    functions []reflect.Value
    addFunc chan<-/*3*/ reflect.Value
    mutex chan/*1*/ bool
    threads int
}

func NewEventLoop() eventLoop {
    var funcs chan reflect.Value
    loop := eventLoop{
        []Reflect.Value{},
        funcs = make(chan reflect.Value, 3),
        make(chan bool, 1),
        0,
    }
    go func(){
        for {
            this.mutex <- 1
            if threads == 0 {
            }
        }
    }
}

func (this eventLoop) NextTick(f func()) {
    this.addFunc <- reflect.ValueOf(f)
}

func (this eventLoop) CurrentTick(f func()) {
    this.mutex <- 1
    threads += 1
    <-this.mutex
    go func() {
        f()
        this.mutex <- 1
        threads -= 1
        <-this.mutex
    }()
}

Answer 1

Если я понимаю ваше намерение, я думаю, вы слишком усложняете вещи.Я бы сделал это так:

package eventloop

type EventLoop struct {
    nextFunc chan func()
    curFunc chan func()
}

func NewEventLoop() *EventLoop {
    el := &EventLoop{
        // Adjust the capacities to taste
        make(chan func(), 3),
        make(chan func(), 3),
    }
    go eventLoop(el)
    return el
}

func (el *EventLoop) NextTick(f func()) {
    el.nextFunc <- f
}

func (el *EventLoop) CurrentTick(f func()) {
    el.curFunc <- f
}

func (el *EventLoop) Quit() {
    close(el.nextFunc)
}

func eventLoop(el *EventLoop) {
    for {
        f, ok := <-el.nextFunc
        if !ok {
            return
        }
        f()

        drain: for {
            select {
            case f := <-el.curFunc:
                f()
            default:
                break drain
            }
        }
    }
}

В зависимости от вашего использования, вам может потребоваться добавить некоторую синхронизацию, чтобы убедиться, что все задачи в цикле завершаются до выхода из вашей программы.

Answer 2

Я понял это сам, после многих проблем и случайных проблем, включая использование длины 15 вместо емкости ... Кажется, у вас есть поток, отправляющий сообщение после уменьшения счетчика.(часть loop.tick может быть встроенной, но я не беспокоюсь об этом)

package eventloop

type eventLoop struct{
    functions []func()
    addFunc chan/*3*/ func()
    mutex chan/*1*/ bool
    threads int
    waitChannel chan bool
    pauseState chan bool
}
func (this *eventLoop) NextTick (f func()) {
    this.addFunc <- f
}

func (this *eventLoop) tick () {
    this.mutex <- true
    for this.threads != 0 {
        <-this.mutex
        <-this.waitChannel
        this.mutex <- true
    }
    <-this.mutex
    L1: for {
        select {
            case f := <-this.addFunc:
                this.functions = append(this.functions,f)
            default: break L1
        }
    }
    if len(this.functions) != 0 {
        this.functions[0]()
        if len(this.functions) >= 2 {
            this.functions = this.functions[1:]
        } else {
            this.functions = []func(){}
        }
    }   else {
        (<-this.addFunc)()
    }
}
func (this *eventLoop) CurrentTick (f func()) {
    this.mutex <- true
    this.threads += 1
    <-this.mutex
    go func() {
        f()
        this.mutex <- true
        this.threads -= 1
        <-this.mutex
        this.waitChannel <- true
    }()
}
func NewEventLoop () *eventLoop {
    funcs := make(chan func(),3)
    loop := &eventLoop{
        make([]func(),0,15), /*functions*/
        funcs, /*addFunc*/
        make(chan bool, 1), /*mutex for threads*/
        0, /*Number of threads*/
        make(chan bool,0), /*The "wait" channel*/
        make(chan bool,1), 
    }
    go func(){
        for { loop.tick() }
    }()
    return loop
}

Примечание: у этого все еще есть много других проблем.