zeromicro/go-zero
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

chore: change interface{} to any (#2818)

Browse Source 
* chore: change interface{} to any

* chore: update goctl version to 1.5.0

* chore: update goctl deps
This commit is contained in:
Kevin Wan
2023-01-24 16:32:02 +08:00
committed by GitHub
parent 7e0ac77139
commit ae87114282
221 changed files with 1910 additions and 2207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

72
core/mr/mapreduce.go
View File

@@ -24,29 +24,29 @@ var (
type (
// ForEachFunc is used to do element processing, but no output.
ForEachFunc func(item interface{})
ForEachFunc func(item any)
// GenerateFunc is used to let callers send elements into source.
GenerateFunc func(source chan<- interface{})
GenerateFunc func(source chan<- any)
// MapFunc is used to do element processing and write the output to writer.
MapFunc func(item interface{}, writer Writer)
MapFunc func(item any, writer Writer)
// MapperFunc is used to do element processing and write the output to writer,
// use cancel func to cancel the processing.
MapperFunc func(item interface{}, writer Writer, cancel func(error))
MapperFunc func(item any, writer Writer, cancel func(error))
// ReducerFunc is used to reduce all the mapping output and write to writer,
// use cancel func to cancel the processing.
ReducerFunc func(pipe <-chan interface{}, writer Writer, cancel func(error))
ReducerFunc func(pipe <-chan any, writer Writer, cancel func(error))
// VoidReducerFunc is used to reduce all the mapping output, but no output.
// Use cancel func to cancel the processing.
VoidReducerFunc func(pipe <-chan interface{}, cancel func(error))
VoidReducerFunc func(pipe <-chan any, cancel func(error))
// Option defines the method to customize the mapreduce.
Option func(opts *mapReduceOptions)
mapperContext struct {
ctx context.Context
mapper MapFunc
source <-chan interface{}
source <-chan any
panicChan *onceChan
collector chan<- interface{}
collector chan<- any
doneChan <-chan lang.PlaceholderType
workers int
}
@@ -58,7 +58,7 @@ type (
// Writer interface wraps Write method.
Writer interface {
Write(v interface{})
Write(v any)
}
)
@@ -68,16 +68,16 @@ func Finish(fns ...func() error) error {
return nil
}
return MapReduceVoid(func(source chan<- interface{}) {
return MapReduceVoid(func(source chan<- any) {
for _, fn := range fns {
source <- fn
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
fn := item.(func() error)
if err := fn(); err != nil {
cancel(err)
}
}, func(pipe <-chan interface{}, cancel func(error)) {
}, func(pipe <-chan any, cancel func(error)) {
}, WithWorkers(len(fns)))
}
@@ -87,11 +87,11 @@ func FinishVoid(fns ...func()) {
return
}
ForEach(func(source chan<- interface{}) {
ForEach(func(source chan<- any) {
for _, fn := range fns {
source <- fn
}
}, func(item interface{}) {
}, func(item any) {
fn := item.(func())
fn()
}, WithWorkers(len(fns)))
@@ -100,14 +100,14 @@ func FinishVoid(fns ...func()) {
// ForEach maps all elements from given generate but no output.
func ForEach(generate GenerateFunc, mapper ForEachFunc, opts ...Option) {
options := buildOptions(opts...)
panicChan := &onceChan{channel: make(chan interface{})}
panicChan := &onceChan{channel: make(chan any)}
source := buildSource(generate, panicChan)
collector := make(chan interface{})
collector := make(chan any)
done := make(chan lang.PlaceholderType)
go executeMappers(mapperContext{
ctx: options.ctx,
mapper: func(item interface{}, _ Writer) {
mapper: func(item any, _ Writer) {
mapper(item)
},
source: source,
@@ -132,25 +132,25 @@ func ForEach(generate GenerateFunc, mapper ForEachFunc, opts ...Option) {
// MapReduce maps all elements generated from given generate func,
// and reduces the output elements with given reducer.
func MapReduce(generate GenerateFunc, mapper MapperFunc, reducer ReducerFunc,
opts ...Option) (interface{}, error) {
panicChan := &onceChan{channel: make(chan interface{})}
opts ...Option) (any, error) {
panicChan := &onceChan{channel: make(chan any)}
source := buildSource(generate, panicChan)
return mapReduceWithPanicChan(source, panicChan, mapper, reducer, opts...)
}
// MapReduceChan maps all elements from source, and reduce the output elements with given reducer.
func MapReduceChan(source <-chan interface{}, mapper MapperFunc, reducer ReducerFunc,
opts ...Option) (interface{}, error) {
panicChan := &onceChan{channel: make(chan interface{})}
func MapReduceChan(source <-chan any, mapper MapperFunc, reducer ReducerFunc,
opts ...Option) (any, error) {
panicChan := &onceChan{channel: make(chan any)}
return mapReduceWithPanicChan(source, panicChan, mapper, reducer, opts...)
}
// mapReduceWithPanicChan maps all elements from source, and reduce the output elements with given reducer.
func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapper MapperFunc,
reducer ReducerFunc, opts ...Option) (interface{}, error) {
func mapReduceWithPanicChan(source <-chan any, panicChan *onceChan, mapper MapperFunc,
reducer ReducerFunc, opts ...Option) (any, error) {
options := buildOptions(opts...)
// output is used to write the final result
output := make(chan interface{})
output := make(chan any)
defer func() {
// reducer can only write once, if more, panic
for range output {
@@ -159,7 +159,7 @@ func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapp
}()
// collector is used to collect data from mapper, and consume in reducer
collector := make(chan interface{}, options.workers)
collector := make(chan any, options.workers)
// if done is closed, all mappers and reducer should stop processing
done := make(chan lang.PlaceholderType)
writer := newGuardedWriter(options.ctx, output, done)
@@ -197,7 +197,7 @@ func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapp
go executeMappers(mapperContext{
ctx: options.ctx,
mapper: func(item interface{}, w Writer) {
mapper: func(item any, w Writer) {
mapper(item, w, cancel)
},
source: source,
@@ -229,7 +229,7 @@ func mapReduceWithPanicChan(source <-chan interface{}, panicChan *onceChan, mapp
// MapReduceVoid maps all elements generated from given generate,
// and reduce the output elements with given reducer.
func MapReduceVoid(generate GenerateFunc, mapper MapperFunc, reducer VoidReducerFunc, opts ...Option) error {
_, err := MapReduce(generate, mapper, func(input <-chan interface{}, writer Writer, cancel func(error)) {
_, err := MapReduce(generate, mapper, func(input <-chan any, writer Writer, cancel func(error)) {
reducer(input, cancel)
}, opts...)
if errors.Is(err, ErrReduceNoOutput) {
@@ -266,8 +266,8 @@ func buildOptions(opts ...Option) *mapReduceOptions {
return options
}
func buildSource(generate GenerateFunc, panicChan *onceChan) chan interface{} {
source := make(chan interface{})
func buildSource(generate GenerateFunc, panicChan *onceChan) chan any {
source := make(chan any)
go func() {
defer func() {
if r := recover(); r != nil {
@@ -283,7 +283,7 @@ func buildSource(generate GenerateFunc, panicChan *onceChan) chan interface{} {
}
// drain drains the channel.
func drain(channel <-chan interface{}) {
func drain(channel <-chan any) {
// drain the channel
for range channel {
}
@@ -348,11 +348,11 @@ func once(fn func(error)) func(error) {
type guardedWriter struct {
ctx context.Context
channel chan<- interface{}
channel chan<- any
done <-chan lang.PlaceholderType
}
func newGuardedWriter(ctx context.Context, channel chan<- interface{},
func newGuardedWriter(ctx context.Context, channel chan<- any,
done <-chan lang.PlaceholderType) guardedWriter {
return guardedWriter{
ctx: ctx,
@@ -361,7 +361,7 @@ func newGuardedWriter(ctx context.Context, channel chan<- interface{},
}
}
func (gw guardedWriter) Write(v interface{}) {
func (gw guardedWriter) Write(v any) {
select {
case <-gw.ctx.Done():
return
@@ -373,11 +373,11 @@ func (gw guardedWriter) Write(v interface{}) {
}
type onceChan struct {
channel chan interface{}
channel chan any
wrote int32
}
func (oc *onceChan) write(val interface{}) {
func (oc *onceChan) write(val any) {
if atomic.CompareAndSwapInt32(&oc.wrote, 0, 1) {
oc.channel <- val
}

8
core/mr/mapreduce_fuzz_test.go
View File

@@ -29,23 +29,23 @@ func FuzzMapReduce(f *testing.F) {
reducerIdx := rand.Int63n(n)
squareSum := (n - 1) * n * (2*n - 1) / 6
fn := func() (interface{}, error) {
fn := func() (any, error) {
defer goleak.VerifyNone(t, goleak.IgnoreCurrent())
return MapReduce(func(source chan<- interface{}) {
return MapReduce(func(source chan<- any) {
for i := int64(0); i < n; i++ {
source <- i
if genPanic && i == genIdx {
panic("foo")
}
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
v := item.(int64)
if mapperPanic && v == mapperIdx {
panic("bar")
}
writer.Write(v * v)
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
var idx int64
var total int64
for v := range pipe {

8
core/mr/mapreduce_fuzzcase_test.go
View File

@@ -54,21 +54,21 @@ func TestMapReduceRandom(t *testing.T) {
reducerIdx := rand.Int63n(n)
squareSum := (n - 1) * n * (2*n - 1) / 6
fn := func() (interface{}, error) {
return MapReduce(func(source chan<- interface{}) {
fn := func() (any, error) {
return MapReduce(func(source chan<- any) {
for i := int64(0); i < n; i++ {
source <- i
if genPanic && i == genIdx {
panic("foo")
}
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
v := item.(int64)
if mapperPanic && v == mapperIdx {
panic("bar")
}
writer.Write(v * v)
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
var idx int64
var total int64
for v := range pipe {

124
core/mr/mapreduce_test.go
View File

@@ -91,11 +91,11 @@ func TestForEach(t *testing.T) {
defer goleak.VerifyNone(t)
var count uint32
ForEach(func(source chan<- interface{}) {
ForEach(func(source chan<- any) {
for i := 0; i < tasks; i++ {
source <- i
}
}, func(item interface{}) {
}, func(item any) {
atomic.AddUint32(&count, 1)
}, WithWorkers(-1))
@@ -106,11 +106,11 @@ func TestForEach(t *testing.T) {
defer goleak.VerifyNone(t)
var count uint32
ForEach(func(source chan<- interface{}) {
ForEach(func(source chan<- any) {
for i := 0; i < tasks; i++ {
source <- i
}
}, func(item interface{}) {
}, func(item any) {
if item.(int)%2 == 0 {
atomic.AddUint32(&count, 1)
}
@@ -123,11 +123,11 @@ func TestForEach(t *testing.T) {
defer goleak.VerifyNone(t)
assert.PanicsWithValue(t, "foo", func() {
ForEach(func(source chan<- interface{}) {
ForEach(func(source chan<- any) {
for i := 0; i < tasks; i++ {
source <- i
}
}, func(item interface{}) {
}, func(item any) {
panic("foo")
})
})
@@ -139,9 +139,9 @@ func TestGeneratePanic(t *testing.T) {
t.Run("all", func(t *testing.T) {
assert.PanicsWithValue(t, "foo", func() {
ForEach(func(source chan<- interface{}) {
ForEach(func(source chan<- any) {
panic("foo")
}, func(item interface{}) {
}, func(item any) {
})
})
})
@@ -154,14 +154,14 @@ func TestMapperPanic(t *testing.T) {
var run int32
t.Run("all", func(t *testing.T) {
assert.PanicsWithValue(t, "foo", func() {
_, _ = MapReduce(func(source chan<- interface{}) {
_, _ = MapReduce(func(source chan<- any) {
for i := 0; i < tasks; i++ {
source <- i
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
atomic.AddInt32(&run, 1)
panic("foo")
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
})
})
assert.True(t, atomic.LoadInt32(&run) < tasks/2)
@@ -176,7 +176,7 @@ func TestMapReduce(t *testing.T) {
mapper MapperFunc
reducer ReducerFunc
expectErr error
expectValue interface{}
expectValue any
}{
{
name: "simple",
@@ -185,7 +185,7 @@ func TestMapReduce(t *testing.T) {
},
{
name: "cancel with error",
mapper: func(item interface{}, writer Writer, cancel func(error)) {
mapper: func(item any, writer Writer, cancel func(error)) {
v := item.(int)
if v%3 == 0 {
cancel(errDummy)
@@ -196,7 +196,7 @@ func TestMapReduce(t *testing.T) {
},
{
name: "cancel with nil",
mapper: func(item interface{}, writer Writer, cancel func(error)) {
mapper: func(item any, writer Writer, cancel func(error)) {
v := item.(int)
if v%3 == 0 {
cancel(nil)
@@ -208,7 +208,7 @@ func TestMapReduce(t *testing.T) {
},
{
name: "cancel with more",
reducer: func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
reducer: func(pipe <-chan any, writer Writer, cancel func(error)) {
var result int
for item := range pipe {
result += item.(int)
@@ -226,13 +226,13 @@ func TestMapReduce(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if test.mapper == nil {
test.mapper = func(item interface{}, writer Writer, cancel func(error)) {
test.mapper = func(item any, writer Writer, cancel func(error)) {
v := item.(int)
writer.Write(v * v)
}
}
if test.reducer == nil {
test.reducer = func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
test.reducer = func(pipe <-chan any, writer Writer, cancel func(error)) {
var result int
for item := range pipe {
result += item.(int)
@@ -240,7 +240,7 @@ func TestMapReduce(t *testing.T) {
writer.Write(result)
}
}
value, err := MapReduce(func(source chan<- interface{}) {
value, err := MapReduce(func(source chan<- any) {
for i := 1; i < 5; i++ {
source <- i
}
@@ -256,13 +256,13 @@ func TestMapReduce(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
if test.mapper == nil {
test.mapper = func(item interface{}, writer Writer, cancel func(error)) {
test.mapper = func(item any, writer Writer, cancel func(error)) {
v := item.(int)
writer.Write(v * v)
}
}
if test.reducer == nil {
test.reducer = func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
test.reducer = func(pipe <-chan any, writer Writer, cancel func(error)) {
var result int
for item := range pipe {
result += item.(int)
@@ -271,7 +271,7 @@ func TestMapReduce(t *testing.T) {
}
}
source := make(chan interface{})
source := make(chan any)
go func() {
for i := 1; i < 5; i++ {
source <- i
@@ -291,13 +291,13 @@ func TestMapReduceWithReduerWriteMoreThanOnce(t *testing.T) {
defer goleak.VerifyNone(t)
assert.Panics(t, func() {
MapReduce(func(source chan<- interface{}) {
MapReduce(func(source chan<- any) {
for i := 0; i < 10; i++ {
source <- i
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
writer.Write(item)
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
drain(pipe)
writer.Write("one")
writer.Write("two")
@@ -323,7 +323,7 @@ func TestMapReduceVoid(t *testing.T) {
},
{
name: "cancel with error",
mapper: func(item interface{}, writer Writer, cancel func(error)) {
mapper: func(item any, writer Writer, cancel func(error)) {
v := item.(int)
if v%3 == 0 {
cancel(errDummy)
@@ -334,7 +334,7 @@ func TestMapReduceVoid(t *testing.T) {
},
{
name: "cancel with nil",
mapper: func(item interface{}, writer Writer, cancel func(error)) {
mapper: func(item any, writer Writer, cancel func(error)) {
v := item.(int)
if v%3 == 0 {
cancel(nil)
@@ -345,7 +345,7 @@ func TestMapReduceVoid(t *testing.T) {
},
{
name: "cancel with more",
reducer: func(pipe <-chan interface{}, cancel func(error)) {
reducer: func(pipe <-chan any, cancel func(error)) {
for item := range pipe {
result := atomic.AddUint32(&value, uint32(item.(int)))
if result > 10 {
@@ -362,19 +362,19 @@ func TestMapReduceVoid(t *testing.T) {
atomic.StoreUint32(&value, 0)
if test.mapper == nil {
test.mapper = func(item interface{}, writer Writer, cancel func(error)) {
test.mapper = func(item any, writer Writer, cancel func(error)) {
v := item.(int)
writer.Write(v * v)
}
}
if test.reducer == nil {
test.reducer = func(pipe <-chan interface{}, cancel func(error)) {
test.reducer = func(pipe <-chan any, cancel func(error)) {
for item := range pipe {
atomic.AddUint32(&value, uint32(item.(int)))
}
}
}
err := MapReduceVoid(func(source chan<- interface{}) {
err := MapReduceVoid(func(source chan<- any) {
for i := 1; i < 5; i++ {
source <- i
}
@@ -392,16 +392,16 @@ func TestMapReduceVoidWithDelay(t *testing.T) {
defer goleak.VerifyNone(t)
var result []int
err := MapReduceVoid(func(source chan<- interface{}) {
err := MapReduceVoid(func(source chan<- any) {
source <- 0
source <- 1
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
i := item.(int)
if i == 0 {
time.Sleep(time.Millisecond * 50)
}
writer.Write(i)
}, func(pipe <-chan interface{}, cancel func(error)) {
}, func(pipe <-chan any, cancel func(error)) {
for item := range pipe {
i := item.(int)
result = append(result, i)
@@ -417,13 +417,13 @@ func TestMapReducePanic(t *testing.T) {
defer goleak.VerifyNone(t)
assert.Panics(t, func() {
_, _ = MapReduce(func(source chan<- interface{}) {
_, _ = MapReduce(func(source chan<- any) {
source <- 0
source <- 1
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
i := item.(int)
writer.Write(i)
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
for range pipe {
panic("panic")
}
@@ -435,17 +435,17 @@ func TestMapReducePanicOnce(t *testing.T) {
defer goleak.VerifyNone(t)
assert.Panics(t, func() {
_, _ = MapReduce(func(source chan<- interface{}) {
_, _ = MapReduce(func(source chan<- any) {
for i := 0; i < 100; i++ {
source <- i
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
i := item.(int)
if i == 0 {
panic("foo")
}
writer.Write(i)
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
for range pipe {
panic("bar")
}
@@ -457,12 +457,12 @@ func TestMapReducePanicBothMapperAndReducer(t *testing.T) {
defer goleak.VerifyNone(t)
assert.Panics(t, func() {
_, _ = MapReduce(func(source chan<- interface{}) {
_, _ = MapReduce(func(source chan<- any) {
source <- 0
source <- 1
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
panic("foo")
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
panic("bar")
})
})
@@ -472,16 +472,16 @@ func TestMapReduceVoidCancel(t *testing.T) {
defer goleak.VerifyNone(t)
var result []int
err := MapReduceVoid(func(source chan<- interface{}) {
err := MapReduceVoid(func(source chan<- any) {
source <- 0
source <- 1
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
i := item.(int)
if i == 1 {
cancel(errors.New("anything"))
}
writer.Write(i)
}, func(pipe <-chan interface{}, cancel func(error)) {
}, func(pipe <-chan any, cancel func(error)) {
for item := range pipe {
i := item.(int)
result = append(result, i)
@@ -496,18 +496,18 @@ func TestMapReduceVoidCancelWithRemains(t *testing.T) {
var done int32
var result []int
err := MapReduceVoid(func(source chan<- interface{}) {
err := MapReduceVoid(func(source chan<- any) {
for i := 0; i < defaultWorkers*2; i++ {
source <- i
}
atomic.AddInt32(&done, 1)
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
i := item.(int)
if i == defaultWorkers/2 {
cancel(errors.New("anything"))
}
writer.Write(i)
}, func(pipe <-chan interface{}, cancel func(error)) {
}, func(pipe <-chan any, cancel func(error)) {
for item := range pipe {
i := item.(int)
result = append(result, i)
@@ -522,13 +522,13 @@ func TestMapReduceWithoutReducerWrite(t *testing.T) {
defer goleak.VerifyNone(t)
uids := []int{1, 2, 3}
res, err := MapReduce(func(source chan<- interface{}) {
res, err := MapReduce(func(source chan<- any) {
for _, uid := range uids {
source <- uid
}
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
writer.Write(item)
}, func(pipe <-chan interface{}, writer Writer, cancel func(error)) {
}, func(pipe <-chan any, writer Writer, cancel func(error)) {
drain(pipe)
// not calling writer.Write(...), should not panic
})
@@ -542,15 +542,15 @@ func TestMapReduceVoidPanicInReducer(t *testing.T) {
const message = "foo"
assert.Panics(t, func() {
var done int32
_ = MapReduceVoid(func(source chan<- interface{}) {
_ = MapReduceVoid(func(source chan<- any) {
for i := 0; i < defaultWorkers*2; i++ {
source <- i
}
atomic.AddInt32(&done, 1)
}, func(item interface{}, writer Writer, cancel func(error)) {
}, func(item any, writer Writer, cancel func(error)) {
i := item.(int)
writer.Write(i)
}, func(pipe <-chan interface{}, cancel func(error)) {
}, func(pipe <-chan any, cancel func(error)) {
panic(message)
}, WithWorkers(1))
})
@@ -561,12 +561,12 @@ func TestForEachWithContext(t *testing.T) {
var done int32
ctx, cancel := context.WithCancel(context.Background())
ForEach(func(source chan<- interface{}) {
ForEach(func(source chan<- any) {
for i := 0; i < defaultWorkers*2; i++ {
source <- i
}
atomic.AddInt32(&done, 1)
}, func(item interface{}) {
}, func(item any) {
i := item.(int)
if i == defaultWorkers/2 {
cancel()
@@ -580,18 +580,18 @@ func TestMapReduceWithContext(t *testing.T) {
var done int32
var result []int
ctx, cancel := context.WithCancel(context.Background())
err := MapReduceVoid(func(source chan<- interface{}) {
err := MapReduceVoid(func(source chan<- any) {
for i := 0; i < defaultWorkers*2; i++ {
source <- i
}
atomic.AddInt32(&done, 1)
}, func(item interface{}, writer Writer, c func(error)) {
}, func(item any, writer Writer, c func(error)) {
i := item.(int)
if i == defaultWorkers/2 {
cancel()
}
writer.Write(i)
}, func(pipe <-chan interface{}, cancel func(error)) {
}, func(pipe <-chan any, cancel func(error)) {
for item := range pipe {
i := item.(int)
result = append(result, i)
@@ -604,10 +604,10 @@ func TestMapReduceWithContext(t *testing.T) {
func BenchmarkMapReduce(b *testing.B) {
b.ReportAllocs()
mapper := func(v interface{}, writer Writer, cancel func(error)) {
mapper := func(v any, writer Writer, cancel func(error)) {
writer.Write(v.(int64) * v.(int64))
}
reducer := func(input <-chan interface{}, writer Writer, cancel func(error)) {
reducer := func(input <-chan any, writer Writer, cancel func(error)) {
var result int64
for v := range input {
result += v.(int64)
@@ -616,7 +616,7 @@ func BenchmarkMapReduce(b *testing.B) {
}
for i := 0; i < b.N; i++ {
MapReduce(func(input chan<- interface{}) {
MapReduce(func(input chan<- any) {
for j := 0; j < 2; j++ {
input <- int64(j)
}

6
core/mr/readme-cn.md
View File

@@ -58,16 +58,16 @@ import (
)
func main() {
val, err := mr.MapReduce(func(source chan<- interface{}) {
val, err := mr.MapReduce(func(source chan<- any) {
// generator
for i := 0; i < 10; i++ {
source <- i
}
}, func(item interface{}, writer mr.Writer, cancel func(error)) {
}, func(item any, writer mr.Writer, cancel func(error)) {
// mapper
i := item.(int)
writer.Write(i * i)
}, func(pipe <-chan interface{}, writer mr.Writer, cancel func(error)) {
}, func(pipe <-chan any, writer mr.Writer, cancel func(error)) {
// reducer
var sum int
for i := range pipe {

6
core/mr/readme.md
View File

@@ -59,16 +59,16 @@ import (
)
func main() {
val, err := mr.MapReduce(func(source chan<- interface{}) {
val, err := mr.MapReduce(func(source chan<- any) {
// generator
for i := 0; i < 10; i++ {
source <- i
}
}, func(item interface{}, writer mr.Writer, cancel func(error)) {
}, func(item any, writer mr.Writer, cancel func(error)) {
// mapper
i := item.(int)
writer.Write(i * i)
}, func(pipe <-chan interface{}, writer mr.Writer, cancel func(error)) {
}, func(pipe <-chan any, writer mr.Writer, cancel func(error)) {
// reducer
var sum int
for i := range pipe {