blob: 3feeff258220ac26f06836180f15397a8cce158f (
plain)
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
|
package runners
import (
"fmt"
"unsafe"
"sync"
)
// Processor acts similarly to a sync.Once object, except that it is reusable. After
// the first call to Process(), any further calls before this first has returned will
// block until the first call has returned, and return the same error. This ensures
// that only a single instance of it is ever running at any one time.
type Processor struct{ p atomic_pointer }
// Process will process the given function if first-call, else blocking until
// the first function has returned, returning the same error result.
func (p *Processor) Process(proc func() error) (err error) {
var i *proc_instance
for {
// Attempt to load existing instance.
ptr := (*proc_instance)(p.p.Load())
if ptr != nil {
// Wait on existing.
ptr.wait.Wait()
err = ptr.err
return
}
if i == nil {
// Allocate instance.
i = new(proc_instance)
i.wait.Add(1)
}
// Try to acquire start slot by
// setting ptr to *our* instance.
if p.p.CAS(nil, unsafe.Pointer(i)) {
defer func() {
if r := recover(); r != nil {
if i.err != nil {
rOld := r // wrap the panic so we don't lose existing returned error
r = fmt.Errorf("panic occured after error %q: %v", i.err.Error(), rOld)
}
// Catch panics and wrap as error return.
i.err = fmt.Errorf("caught panic: %v", r)
}
// Set return.
err = i.err
// Release the
// goroutines.
i.wait.Done()
// Free processor.
p.p.Store(nil)
}()
// Run func.
i.err = proc()
return
}
}
}
type proc_instance struct {
wait sync.WaitGroup
err error
}
|
