关于golang:聊聊storagetapper的pool

1次阅读

共计 2934 个字符，预计需要花费 8 分钟才能阅读完成。

本文次要钻研一下 storagetapper 的 pool

storagetapper/pool/pool.go

type Thread interface {Start(m uint, f func())
    Adjust(m uint)
    Terminate() bool
    NumProcs() uint}

Thread 接口定义了 Start、Adjust、Terminate、NumProcs 办法

storagetapper/pool/pool.go

/*Create helps to hide poolImpl in the package, but not really required */
func Create() Thread {return &poolImpl{}
}

type poolImpl struct {
    mutex       sync.Mutex
    numProcs    uint
    maxNumProcs uint
    fn          func()}

/*Start instantiates a pool of size of 'm' of 'f' goroutines */
/*Start and Create separation allows to pass pool instance to 'f' goroutine */
func (p *poolImpl) Start(m uint, f func()) {
    p.fn = f
    p.Adjust(m)
}

/*Adjust resizes the pool. It creates new threads if requested size is bigger
* then current size, while it assumes threads cooperation when requested size is
* smaller then current size. Threads should periodically call Terminate function
* and obey the result. */
func (p *poolImpl) Adjust(m uint) {p.mutex.Lock()
    defer p.mutex.Unlock()
    log.Debugf("Current size=%v, current maximum size=%v, requested size=%v", p.numProcs, p.maxNumProcs, m)
    p.maxNumProcs = m
    if p.numProcs < p.maxNumProcs {
        adj := p.maxNumProcs - p.numProcs
        shutdown.Register(int32(adj))
        for i := uint(0); i < adj; i++ {go func() {defer shutdown.Done(); p.fn()}()}
        p.numProcs = m
    }
}

/*Terminate return true if the caller thread need to terminate */
func (p *poolImpl) Terminate() bool {
    //Uncomment if Terminate is called frequently
    //Introduces a race when thread can miss Pool resize event, that's ok, so as
    //some other threads may see the event, or we will see it on the next
    //iteration
    //    if p.numProcs <= p.maxNumProcs {
    //        return false
    //    }

    p.mutex.Lock()
    defer p.mutex.Unlock()

    if p.numProcs > p.maxNumProcs {
        p.numProcs--
        log.Debugf("Terminating. Current size=%v, current maximum size=%v", p.numProcs, p.maxNumProcs)
        return true
    }

    return false
}

/*NumProcs return current size of the pool */
func (p *poolImpl) NumProcs() uint {p.mutex.Lock()
    defer p.mutex.Unlock()
    return p.numProcs
}

poolImpl 定义了 mutex、numProcs、maxNumProcs、fn 属性；它实现了 Thread 接口，其 Start 办法设置了 fn，同时执行 Adjust 办法；Adjust 办法在 numProcs 小于 maxNumProcs 时会执行 shutdown.Register，而后挨个执行 shutdown.Done()；Terminate 办法对于 numProcs 大于 maxNumProcs 的状况递加 numProcs

storagetapper/pool/pool_test.go

func TestBasic(t *testing.T) {
    var m sync.Mutex
    var nProcs int32

    sig := make(chan bool)

    p := Create()

    if p.NumProcs() != 0 {t.Fatalf("Initially not zero")
    }

    p.Start(2, func() {m.Lock()
        atomic.AddInt32(&nProcs, 1)
        log.Debugf("Starting new proc, nProcs=%v", nProcs)
        m.Unlock()
        for !p.Terminate() {
            <-sig
            log.Debugf("Woken up")
        }
        m.Lock()
        atomic.AddInt32(&nProcs, -1)
        log.Debugf("Terminating proc, nProcs=%v", nProcs)
        m.Unlock()})

    /* Check that both real number and reported by thread pool equal to expected
    * value */
    waitFor(&nProcs, 2, 5, t)
    if p.NumProcs() != 2 {t.Fatalf("numProcs != 2")
    }

    p.Adjust(8)

    waitFor(&nProcs, 8, 5, t)
    if p.NumProcs() != 8 {t.Fatalf("numProcs != 8")
    }

    p.Adjust(3)

    for i := 0; i < 5; i++ {sig <- true}

    waitFor(&nProcs, 3, 5, t)
    if p.NumProcs() != 3 {t.Fatalf("numProcs != 3")
    }

    p.Adjust(0)
    for i := 0; i < 3; i++ {sig <- true}

    waitFor(&nProcs, 0, 5, t)
    if p.NumProcs() != 0 {t.Fatalf("numProcs != 0")
    }
}

storagetapper 的 Thread 接口定义了 Start、Adjust、Terminate、NumProcs 办法；poolImpl 实现了 Thread 接口；其 Adjust 能够在 numProcs 小于 maxNumProcs 的时候进行扩容；Terminate 会在 numProcs 大于 maxNumProcs 的时候递加 numProcs。