test: add more benchmark (#34)

Author: siyul-park

pkg/loader/reconciler.go

@@ -7,35 +7,29 @@ import (
 	"github.com/siyul-park/uniflow/pkg/storage"
 )
 
-type (
-	// ReconcilerConfig holds the configuration for the Reconciler.
-	ReconcilerConfig struct {
-		Storage *storage.Storage // Storage is the storage used by the Reconciler.
-		Loader  *Loader          // Loader is used to load scheme.Spec into the symbol.Table.
-		Filter  *storage.Filter  // Filter is the filter for tracking changes to the scheme.Spec.
-	}
+// ReconcilerConfig holds the configuration for the Reconciler.
+type ReconcilerConfig struct {
+	Storage *storage.Storage // Storage is the storage used by the Reconciler.
+	Loader  *Loader          // Loader is used to load scheme.Spec into the symbol.Table.
+	Filter  *storage.Filter  // Filter is the filter for tracking changes to the scheme.Spec.
+}
 
-	// Reconciler keeps the symbol.Table up to date by tracking changes to scheme.Spec.
-	Reconciler struct {
-		storage *storage.Storage
-		loader  *Loader
-		filter  *storage.Filter
-		stream  *storage.Stream
-		done    chan struct{}
-		mu      sync.Mutex
-	}
-)
+// Reconciler keeps the symbol.Table up to date by tracking changes to scheme.Spec.
+type Reconciler struct {
+	storage *storage.Storage
+	loader  *Loader
+	filter  *storage.Filter
+	stream  *storage.Stream
+	done    chan struct{}
+	mu      sync.Mutex
+}
 
 // NewReconciler creates a new Reconciler with the given configuration.
 func NewReconciler(config ReconcilerConfig) *Reconciler {
-	storage := config.Storage
-	loader := config.Loader
-	filter := config.Filter
-
 	return &Reconciler{
-		storage: storage,
-		loader:  loader,
-		filter:  filter,
+		storage: config.Storage,
+		loader:  config.Loader,
+		filter:  config.Filter,
 		done:    make(chan struct{}),
 	}
 }

pkg/node/error.go

@@ -5,11 +5,16 @@ import (
 	"github.com/siyul-park/uniflow/pkg/port"
 )
 
-type (
-	// Node is an operational unit that processes *packet.Packet.
-	Node interface {
-		ID() ulid.ULID
-		Port(name string) (*port.Port, bool)
-		Close() error
-	}
+// Node is an operational unit that processes *packet.Packet.
+type Node interface {
+	ID() ulid.ULID
+	Port(name string) (*port.Port, bool)
+	Close() error
+}
+
+const (
+	PortIO  = "io"
+	PortIn  = "in"
+	PortOut = "out"
+	PortErr = "error"
 )

pkg/node/node.go

@@ -9,27 +9,25 @@ import (
 	"github.com/siyul-park/uniflow/pkg/process"
 )
 
-type (
-	// OneToManyNodeConfig is a config for ActionNode.
-	OneToManyNodeConfig struct {
-		ID     ulid.ULID
-		Action func(*process.Process, *packet.Packet) ([]*packet.Packet, *packet.Packet)
-	}
+// OneToManyNodeConfig holds the configuration for OneToManyNode.
+type OneToManyNodeConfig struct {
+	ID     ulid.ULID
+	Action func(*process.Process, *packet.Packet) ([]*packet.Packet, *packet.Packet)
+}
 
-	// OneToManyNode provide process *packet.Packet one source and many distance.
-	OneToManyNode struct {
-		id       ulid.ULID
-		action   func(*process.Process, *packet.Packet) ([]*packet.Packet, *packet.Packet)
-		inPort   *port.Port
-		outPorts []*port.Port
-		errPort  *port.Port
-		mu       sync.RWMutex
-	}
-)
+// OneToManyNode represents a node that processes *packet.Packet with one input and many outputs.
+type OneToManyNode struct {
+	id       ulid.ULID
+	action   func(*process.Process, *packet.Packet) ([]*packet.Packet, *packet.Packet)
+	inPort   *port.Port
+	outPorts []*port.Port
+	errPort  *port.Port
+	mu       sync.RWMutex
+}
 
 var _ Node = (*OneToManyNode)(nil)
 
-// NewOneToManyNode returns a new OneToManyNode.
+// NewOneToManyNode creates a new OneToManyNode with the given configuration.
 func NewOneToManyNode(config OneToManyNodeConfig) *OneToManyNode {
 	id := config.ID
 	action := config.Action
@@ -64,13 +62,15 @@ func NewOneToManyNode(config OneToManyNodeConfig) *OneToManyNode {
 	return n
 }
 
+// ID returns the ID of the OneToManyNode.
 func (n *OneToManyNode) ID() ulid.ULID {
 	n.mu.RLock()
 	defer n.mu.RUnlock()
 
 	return n.id
 }
 
+// Port returns the specified port of the OneToManyNode.
 func (n *OneToManyNode) Port(name string) (*port.Port, bool) {
 	n.mu.Lock()
 	defer n.mu.Unlock()
@@ -81,30 +81,30 @@ func (n *OneToManyNode) Port(name string) (*port.Port, bool) {
 	case PortErr:
 		return n.errPort, true
 	default:
-	}
-
-	if i, ok := port.GetIndex(PortOut, name); ok {
-		for j := 0; j <= i; j++ {
-			if len(n.outPorts) <= j {
-				outPort := port.New()
-				outPort.AddInitHook(port.InitHookFunc(func(proc *process.Process) {
-					n.mu.RLock()
-					defer n.mu.RUnlock()
-
-					outStream := outPort.Open(proc)
-
-					n.backward(proc, outStream)
-				}))
-				n.outPorts = append(n.outPorts, outPort)
+		if i, ok := port.GetIndex(PortOut, name); ok {
+			for j := 0; j <= i; j++ {
+				if len(n.outPorts) <= j {
+					outPort := port.New()
+					outPort.AddInitHook(port.InitHookFunc(func(proc *process.Process) {
+						n.mu.RLock()
+						defer n.mu.RUnlock()
+
+						outStream := outPort.Open(proc)
+
+						n.backward(proc, outStream)
+					}))
+					n.outPorts = append(n.outPorts, outPort)
+				}
 			}
-		}
 
-		return n.outPorts[i], true
+			return n.outPorts[i], true
+		}
 	}
 
 	return nil, false
 }
 
+// Close closes all ports of the OneToManyNode.
 func (n *OneToManyNode) Close() error {
 	n.mu.Lock()
 	defer n.mu.Unlock()
@@ -147,30 +147,21 @@ func (n *OneToManyNode) forward(proc *process.Process) {
 				inStream.Send(errPck)
 			}
 		} else if len(outPcks) > 0 && len(outStreams) > 0 {
-			var ok bool
 			for i, outPck := range outPcks {
-				if len(outStreams) <= i {
-					break
-				}
-				if outPck == nil {
+				if len(outStreams) <= i || outPck == nil {
 					continue
 				}
 				outStream := outStreams[i]
-
-				if outStream.Links() > 0 {
-					if outPck == inPck {
-						outPck = packet.New(outPck.Payload())
-					}
-					proc.Stack().Link(inPck.ID(), outPck.ID())
-					proc.Stack().Push(outPck.ID(), inStream.ID())
-					outStream.Send(outPck)
-
-					ok = true
+				if outStream.Links() == 0 {
+					continue
 				}
-			}
 
-			if !ok {
-				proc.Stack().Clear(inPck.ID())
+				if outPck == inPck {
+					outPck = packet.New(outPck.Payload())
+				}
+				proc.Stack().Link(inPck.ID(), outPck.ID())
+				proc.Stack().Push(outPck.ID(), inStream.ID())
+				outStream.Send(outPck)
 			}
 		} else {
 			proc.Stack().Clear(inPck.ID())

pkg/node/onetomany.go

@@ -130,3 +130,36 @@ func TestOneToManyNode_Send(t *testing.T) {
 		}
 	})
 }
+
+func BenchmarkOneToManyNode_Send(b *testing.B) {
+	n := NewOneToManyNode(OneToManyNodeConfig{
+		Action: func(_ *process.Process, inPck *packet.Packet) ([]*packet.Packet, *packet.Packet) {
+			return []*packet.Packet{inPck}, nil
+		},
+	})
+	defer func() { _ = n.Close() }()
+
+	in := port.New()
+	inPort, _ := n.Port(PortIn)
+	inPort.Link(in)
+
+	out := port.New()
+	outPort, _ := n.Port(port.SetIndex(PortOut, 0))
+	outPort.Link(out)
+
+	proc := process.New()
+	defer proc.Exit(nil)
+
+	inStream := in.Open(proc)
+	outStream := out.Open(proc)
+
+	inPayload := primitive.NewString(faker.UUIDHyphenated())
+	inPck := packet.New(inPayload)
+
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		inStream.Send(inPck)
+		<-outStream.Receive()
+	}
+}

pkg/node/onetomany_test.go

@@ -9,28 +9,26 @@ import (
 	"github.com/siyul-park/uniflow/pkg/process"
 )
 
-type (
-	// OneToOneNodeConfig is a config for ActionNode.
-	OneToOneNodeConfig struct {
-		ID     ulid.ULID
-		Action func(*process.Process, *packet.Packet) (*packet.Packet, *packet.Packet)
-	}
+// OneToOneNodeConfig holds the configuration for OneToOneNode.
+type OneToOneNodeConfig struct {
+	ID     ulid.ULID
+	Action func(*process.Process, *packet.Packet) (*packet.Packet, *packet.Packet)
+}
 
-	// OneToOneNode provide process *packet.Packet one source and onde distance.
-	OneToOneNode struct {
-		id      ulid.ULID
-		action  func(*process.Process, *packet.Packet) (*packet.Packet, *packet.Packet)
-		ioPort  *port.Port
-		inPort  *port.Port
-		outPort *port.Port
-		errPort *port.Port
-		mu      sync.RWMutex
-	}
-)
+// OneToOneNode represents a node that processes *packet.Packet with one input and one output.
+type OneToOneNode struct {
+	id      ulid.ULID
+	action  func(*process.Process, *packet.Packet) (*packet.Packet, *packet.Packet)
+	ioPort  *port.Port
+	inPort  *port.Port
+	outPort *port.Port
+	errPort *port.Port
+	mu      sync.RWMutex
+}
 
 var _ Node = (*OneToOneNode)(nil)
 
-// NewOneToOneNode returns a new OneToOneNode.
+// NewOneToOneNode creates a new OneToOneNode with the given configuration.
 func NewOneToOneNode(config OneToOneNodeConfig) *OneToOneNode {
 	id := config.ID
 	action := config.Action
@@ -90,13 +88,15 @@ func NewOneToOneNode(config OneToOneNodeConfig) *OneToOneNode {
 	return n
 }
 
+// ID returns the ID of the OneToOneNode.
 func (n *OneToOneNode) ID() ulid.ULID {
 	n.mu.RLock()
 	defer n.mu.RUnlock()
 
 	return n.id
 }
 
+// Port returns the specified port of the OneToOneNode.
 func (n *OneToOneNode) Port(name string) (*port.Port, bool) {
 	n.mu.RLock()
 	defer n.mu.RUnlock()
@@ -116,6 +116,7 @@ func (n *OneToOneNode) Port(name string) (*port.Port, bool) {
 	return nil, false
 }
 
+// Close closes all ports of the OneToOneNode.
 func (n *OneToOneNode) Close() error {
 	n.mu.Lock()
 	defer n.mu.Unlock()
@@ -128,7 +129,7 @@ func (n *OneToOneNode) Close() error {
 	return nil
 }
 
-func (n *OneToOneNode) forward(proc *process.Process, inStream *port.Stream, outStream *port.Stream) {
+func (n *OneToOneNode) forward(proc *process.Process, inStream, outStream *port.Stream) {
 	n.mu.RLock()
 	defer n.mu.RUnlock()