当 GPT-4 拥有了 Diff 视图，那真的是如虎添翼！- 每天5分钟玩转 GPT 编程系列(9)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

package gopool

import (
	"context"
	"sync"
	"time"
)

// GoPool represents a pool of workers.
type GoPool interface {
	// AddTask adds a task to the pool.
	AddTask(t task)
	// Wait waits for all tasks to be dispatched and completed.
	Wait()
	// Release releases the pool and all its workers.
	Release()
	// GetRunning returns the number of running workers.
	Running() int
	// GetWorkerCount returns the number of workers.
	GetWorkerCount() int
}

// task represents a function that will be executed by a worker.
// It returns a result and an error.
type task func() (interface{}, error)

// goPool represents a pool of workers.
type goPool struct {
	workers     []*worker
	workerStack []int
	maxWorkers  int
	// Set by WithMinWorkers(), used to adjust the number of workers. Default equals to maxWorkers.
	minWorkers int
	// tasks are added to this channel first, then dispatched to workers. Default buffer size is 1 million.
	taskQueue chan task
	// Set by WithTaskQueueSize(), used to set the size of the task queue. Default is 1e6.
	taskQueueSize int
	// Set by WithRetryCount(), used to retry a task when it fails. Default is 0.
	retryCount int
	lock       sync.Locker
	cond       *sync.Cond
	// Set by WithTimeout(), used to set a timeout for a task. Default is 0, which means no timeout.
	timeout time.Duration
	// Set by WithResultCallback(), used to handle the result of a task. Default is nil.
	resultCallback func(interface{})
	// Set by WithErrorCallback(), used to handle the error of a task. Default is nil.
	errorCallback func(error)
	// adjustInterval is the interval to adjust the number of workers. Default is 1 second.
	adjustInterval time.Duration
	ctx            context.Context
	// cancel is used to cancel the context. It is called when Release() is called.
	cancel context.CancelFunc
}

// NewGoPool creates a new pool of workers.
func NewGoPool(maxWorkers int, opts ...Option) GoPool {
	ctx, cancel := context.WithCancel(context.Background())
	pool := &goPool{
		maxWorkers: maxWorkers,
		// Set minWorkers to maxWorkers by default
		minWorkers: maxWorkers,
		// workers and workerStack should be initialized after WithMinWorkers() is called
		workers:        nil,
		workerStack:    nil,
		taskQueue:      nil,
		taskQueueSize:  1e6,
		retryCount:     0,
		lock:           new(sync.Mutex),
		timeout:        0,
		adjustInterval: 1 * time.Second,
		ctx:            ctx,
		cancel:         cancel,
	}
	// Apply options
	for _, opt := range opts {
		opt(pool)
	}

	pool.taskQueue = make(chan task, pool.taskQueueSize)

	pool.workers = make([]*worker, pool.minWorkers)
	pool.workerStack = make([]int, pool.minWorkers)

	if pool.cond == nil {
		pool.cond = sync.NewCond(pool.lock)
	}
	// Create workers with the minimum number. Don't use pushWorker() here.
	for i := 0; i < pool.minWorkers; i++ {
		worker := newWorker()
		pool.workers[i] = worker
		pool.workerStack[i] = i
		worker.start(pool, i)
	}
	go pool.adjustWorkers()
	go pool.dispatch()
	return pool
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

package gopool

import (
	"sync"
	"time"
)

// Option represents an option for the pool.
type Option func(*goPool)

// WithTaskQueueSize sets the size of the task queue for the pool.
func WithTaskQueueSize(size int) Option {
	return func(p *goPool) {
		p.taskQueueSize = size
	}
}

1

pool := gopool.NewGoPool(100, gopool.WithTaskQueueSize(5000))

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

package main

import (
    "sync"
    "time"

    "github.com/devchat-ai/gopool"
)

func main() {
    pool := gopool.NewGoPool(100)
    defer pool.Release()

    for i := 0; i < 1000; i++ {
        pool.AddTask(func() (interface{}, error){
            time.Sleep(10 * time.Millisecond)
            return nil, nil
        })
    }
    pool.Wait()
}

1
2
3
4
5
6
7

type goPool struct {
	workers     []*worker
    // ......
	// tasks are added to this channel first, then dispatched to workers. Default buffer size is 1 million.
	taskQueue chan task
    // ......
}