diff --git a/pkg/reducer/task_manager.go b/pkg/reducer/task_manager.go index a722eba3..537f6b59 100644 --- a/pkg/reducer/task_manager.go +++ b/pkg/reducer/task_manager.go @@ -95,8 +95,6 @@ func (rtm *reduceTaskManager) CreateTask(ctx context.Context, request *v1.Reduce // write the output to the output channel, service will forward it to downstream rtm.responseCh <- task.buildReduceResponse(message) } - // send EOF - rtm.responseCh <- task.buildEOFResponse() // close the output channel after the reduce function is done close(task.outputCh) // send a done signal @@ -133,27 +131,21 @@ func (rtm *reduceTaskManager) OutputChannel() <-chan *v1.ReduceResponse { // WaitAll waits for all the reduce tasks to complete. func (rtm *reduceTaskManager) WaitAll() { - tasks := make([]*reduceTask, 0, len(rtm.tasks)) + var eofResponse *v1.ReduceResponse for _, task := range rtm.tasks { - tasks = append(tasks, task) - } - - for _, task := range tasks { <-task.doneCh + if eofResponse == nil { + eofResponse = task.buildEOFResponse() + } } - + rtm.responseCh <- eofResponse // after all the tasks are completed, close the output channel close(rtm.responseCh) } // CloseAll closes all the reduce tasks. func (rtm *reduceTaskManager) CloseAll() { - tasks := make([]*reduceTask, 0, len(rtm.tasks)) for _, task := range rtm.tasks { - tasks = append(tasks, task) - } - - for _, task := range tasks { close(task.inputCh) } } diff --git a/pkg/reducestreamer/task_manager.go b/pkg/reducestreamer/task_manager.go index ae04f096..15c2cc62 100644 --- a/pkg/reducestreamer/task_manager.go +++ b/pkg/reducestreamer/task_manager.go @@ -95,8 +95,6 @@ func (rtm *reduceStreamTaskManager) CreateTask(ctx context.Context, request *v1. // write the output to the output channel, service will forward it to downstream rtm.responseCh <- task.buildReduceResponse(message) } - // send EOF - rtm.responseCh <- task.buildEOFResponse() }() reduceStreamerHandle := rtm.creatorHandle.Create() @@ -141,26 +139,22 @@ func (rtm *reduceStreamTaskManager) OutputChannel() <-chan *v1.ReduceResponse { // WaitAll waits for all the reduceStream tasks to complete. func (rtm *reduceStreamTaskManager) WaitAll() { - tasks := make([]*reduceStreamTask, 0, len(rtm.tasks)) + var eofResponse *v1.ReduceResponse for _, task := range rtm.tasks { - tasks = append(tasks, task) - } - - for _, task := range tasks { <-task.doneCh + if eofResponse == nil { + eofResponse = task.buildEOFResponse() + } } + rtm.responseCh <- eofResponse + // after all the tasks are completed, close the output channel close(rtm.responseCh) } // CloseAll closes all the reduceStream tasks. func (rtm *reduceStreamTaskManager) CloseAll() { - tasks := make([]*reduceStreamTask, 0, len(rtm.tasks)) for _, task := range rtm.tasks { - tasks = append(tasks, task) - } - - for _, task := range tasks { close(task.inputCh) } } diff --git a/pkg/sessionreducer/examples/sum/Dockerfile b/pkg/sessionreducer/examples/sum/Dockerfile new file mode 100644 index 00000000..4fd55692 --- /dev/null +++ b/pkg/sessionreducer/examples/sum/Dockerfile @@ -0,0 +1,20 @@ +#################################################################################################### +# base +#################################################################################################### +FROM alpine:3.12.3 as base +RUN apk update && apk upgrade && \ + apk add ca-certificates && \ + apk --no-cache add tzdata + +COPY dist/sum-example /bin/sum-example +RUN chmod +x /bin/sum-example + +#################################################################################################### +# counter +#################################################################################################### +FROM scratch as sum +ARG ARCH +COPY --from=base /usr/share/zoneinfo /usr/share/zoneinfo +COPY --from=base /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt +COPY --from=base /bin/sum-example /bin/sum-example +ENTRYPOINT [ "/bin/sum-example" ] diff --git a/pkg/sessionreducer/examples/sum/Makefile b/pkg/sessionreducer/examples/sum/Makefile new file mode 100644 index 00000000..e421ef11 --- /dev/null +++ b/pkg/sessionreducer/examples/sum/Makefile @@ -0,0 +1,18 @@ +TAG ?= stable +PUSH ?= false + +.PHONY: build +build: + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build -v -o ./dist/sum-example main.go + +.PHONY: image-push +image-push: build + docker buildx build -t "quay.io/numaio/numaflow-go/session-sum:${TAG}" --platform linux/amd64,linux/arm64 --target sum . --push + +.PHONY: image +image: build + docker build -t "quay.io/numaio/numaflow-go/session-sum:${TAG}" --target sum . + @if [ "$(PUSH)" = "true" ]; then docker push "quay.io/numaio/numaflow-go/session-sum:${TAG}"; fi + +clean: + -rm -rf ./dist diff --git a/pkg/sessionreducer/examples/sum/README.md b/pkg/sessionreducer/examples/sum/README.md new file mode 100644 index 00000000..e17d7cad --- /dev/null +++ b/pkg/sessionreducer/examples/sum/README.md @@ -0,0 +1,3 @@ +# Sum + +An example User Defined Function that computes sum of events. diff --git a/pkg/sessionreducer/examples/sum/go.mod b/pkg/sessionreducer/examples/sum/go.mod new file mode 100644 index 00000000..c5c31e26 --- /dev/null +++ b/pkg/sessionreducer/examples/sum/go.mod @@ -0,0 +1,21 @@ +module counter + +go 1.20 + +replace github.com/numaproj/numaflow-go => ../../../.. + +require ( + github.com/numaproj/numaflow-go v0.6.1-0.20240212202512-715955f1e069 + go.uber.org/atomic v1.11.0 +) + +require ( + github.com/golang/protobuf v1.5.3 // indirect + golang.org/x/net v0.9.0 // indirect + golang.org/x/sync v0.1.0 // indirect + golang.org/x/sys v0.7.0 // indirect + golang.org/x/text v0.9.0 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 // indirect + google.golang.org/grpc v1.57.0 // indirect + google.golang.org/protobuf v1.31.0 // indirect +) diff --git a/pkg/sessionreducer/examples/sum/go.sum b/pkg/sessionreducer/examples/sum/go.sum new file mode 100644 index 00000000..ae935bd8 --- /dev/null +++ b/pkg/sessionreducer/examples/sum/go.sum @@ -0,0 +1,32 @@ +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.3 h1:KhyjKVUg7Usr/dYsdSqoFveMYd5ko72D+zANwlG1mmg= +github.com/golang/protobuf v1.5.3/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38= +github.com/numaproj/numaflow-go v0.6.1-0.20240212202512-715955f1e069 h1:LBMIrUytoxuZazEnwxo+WPAxjHL1rrVhphTyByLiDYY= +github.com/numaproj/numaflow-go v0.6.1-0.20240212202512-715955f1e069/go.mod h1:WoMt31+h3up202zTRI8c/qe42B8UbvwLe2mJH0MAlhI= +github.com/numaproj/numaflow-go v0.6.1-0.20240221033908-7c1198bc89f1 h1:yY1ZjhfqdxE9H4T9YRff+GJHilv9CJgslWEYfsE/EPo= +github.com/numaproj/numaflow-go v0.6.1-0.20240221033908-7c1198bc89f1/go.mod h1:WoMt31+h3up202zTRI8c/qe42B8UbvwLe2mJH0MAlhI= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE= +go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= +golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM= +golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns= +golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o= +golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU= +golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE= +golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19 h1:0nDDozoAU19Qb2HwhXadU8OcsiO/09cnTqhUtq2MEOM= +google.golang.org/genproto/googleapis/rpc v0.0.0-20230525234030-28d5490b6b19/go.mod h1:66JfowdXAEgad5O9NnYcsNPLCPZJD++2L9X0PCMODrA= +google.golang.org/grpc v1.57.0 h1:kfzNeI/klCGD2YPMUlaGNT3pxvYfga7smW3Vth8Zsiw= +google.golang.org/grpc v1.57.0/go.mod h1:Sd+9RMTACXwmub0zcNY2c4arhtrbBYD1AUHI/dt16Mo= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.31.0 h1:g0LDEJHgrBl9N9r17Ru3sqWhkIx2NB67okBHPwC7hs8= +google.golang.org/protobuf v1.31.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/pkg/sessionreducer/examples/sum/main.go b/pkg/sessionreducer/examples/sum/main.go new file mode 100644 index 00000000..f6398211 --- /dev/null +++ b/pkg/sessionreducer/examples/sum/main.go @@ -0,0 +1,59 @@ +package main + +import ( + "context" + "fmt" + "log" + "strconv" + + "go.uber.org/atomic" + + "github.com/numaproj/numaflow-go/pkg/sessionreducer" +) + +// Sum is a simple session reducer which computes sum of events in a session. +type Sum struct { + sum *atomic.Int32 +} + +func (c *Sum) SessionReduce(ctx context.Context, keys []string, input <-chan sessionreducer.Datum, outputCh chan<- sessionreducer.Message) { + for d := range input { + val, err := strconv.Atoi(string(d.Value())) + if err != nil { + log.Panic("unable to convert the value to int: ", err.Error()) + } else { + c.sum.Add(int32(val)) + } + } + outputCh <- sessionreducer.NewMessage([]byte(fmt.Sprintf("%d", c.sum.Load()))).WithKeys(keys) +} + +func (c *Sum) Accumulator(ctx context.Context) []byte { + return []byte(strconv.Itoa(int(c.sum.Load()))) +} + +func (c *Sum) MergeAccumulator(ctx context.Context, accumulator []byte) { + val, err := strconv.Atoi(string(accumulator)) + if err != nil { + log.Println("unable to convert the accumulator value to int: ", err.Error()) + return + } + c.sum.Add(int32(val)) +} + +func NewSessionCounter() sessionreducer.SessionReducer { + return &Sum{ + sum: atomic.NewInt32(0), + } +} + +// SessionCounterCreator is the creator for the session reducer. +type SessionCounterCreator struct{} + +func (s *SessionCounterCreator) Create() sessionreducer.SessionReducer { + return NewSessionCounter() +} + +func main() { + sessionreducer.NewServer(&SessionCounterCreator{}).Start(context.Background()) +}