tcp: optimize memory for observatio

Author: jkralik

message/options.go

@@ -550,3 +550,18 @@ func (options Options) ResetOptionsTo(buf []byte, in Options) (Options, int, err
 	}
 	return opts, used, nil
 }
+
+// Clone create duplicates of options.
+func (options Options) Clone() (Options, error) {
+	opts := make(Options, 0, len(options))
+	buf := make([]byte, 64)
+	opts, used, err := opts.ResetOptionsTo(buf, options)
+	if err == ErrTooSmall {
+		buf = append(buf, make([]byte, used-len(buf))...)
+		opts, used, err = opts.ResetOptionsTo(buf, options)
+	}
+	if err != nil {
+		return nil, err
+	}
+	return opts, nil
+}

tcp/clientconn.go

@@ -106,14 +106,14 @@ func bwReleaseMessage(m blockwise.Message) {
 	pool.ReleaseMessage(m.(*pool.Message))
 }
 
-func bwCreateHandlerFunc(observatioRequests *kitSync.Map) func(token message.Token) (blockwise.Message, bool) {
+func bwCreateHandlerFunc(observationRequests *kitSync.Map) func(token message.Token) (blockwise.Message, bool) {
 	return func(token message.Token) (blockwise.Message, bool) {
-		msg, ok := observatioRequests.LoadWithFunc(token.String(), func(v interface{}) interface{} {
-			r := v.(*pool.Message)
-			d := pool.AcquireMessage(r.Context())
-			d.ResetOptionsTo(r.Options())
-			d.SetCode(r.Code())
-			d.SetToken(r.Token())
+		msg, ok := observationRequests.LoadWithFunc(token.String(), func(v interface{}) interface{} {
+			r := v.(message.Message)
+			d := pool.AcquireMessage(r.Context)
+			d.ResetOptionsTo(r.Options)
+			d.SetCode(r.Code)
+			d.SetToken(r.Token)
 			return d
 		})
 		if !ok {
@@ -134,7 +134,7 @@ func Client(conn net.Conn, opts ...DialOption) *ClientConn {
 		cfg.errors = func(error) {}
 	}
 
-	observatioRequests := kitSync.NewMap()
+	observationRequests := kitSync.NewMap()
 	var blockWise *blockwise.BlockWise
 	if cfg.blockwiseEnable {
 		blockWise = blockwise.NewBlockWise(
@@ -143,7 +143,7 @@ func Client(conn net.Conn, opts ...DialOption) *ClientConn {
 			cfg.blockwiseTransferTimeout,
 			cfg.errors,
 			false,
-			bwCreateHandlerFunc(observatioRequests),
+			bwCreateHandlerFunc(observationRequests),
 		)
 	}
 
@@ -162,7 +162,7 @@ func Client(conn net.Conn, opts ...DialOption) *ClientConn {
 		cfg.disableTCPSignalMessageCSM,
 		cfg.closeSocket,
 	)
-	cc := NewClientConn(session, observationTokenHandler, observatioRequests)
+	cc := NewClientConn(session, observationTokenHandler, observationRequests)
 
 	go func() {
 		err := cc.Run()

tcp/clientobserve.go

@@ -38,10 +38,7 @@ func NewObservationHandler(obsertionTokenHandler *HandlerContainer, next Handler
 
 func (o *Observation) cleanUp() {
 	o.cc.observationTokenHandler.Pop(o.token)
-	registeredRequest, ok := o.cc.observationRequests.PullOut(o.token.String())
-	if ok {
-		pool.ReleaseMessage(registeredRequest.(*pool.Message))
-	}
+	o.cc.observationRequests.PullOut(o.token.String())
 }
 
 // Cancel remove observation from server. For recreate observation use Observe.
@@ -89,6 +86,7 @@ func (cc *ClientConn) Observe(ctx context.Context, path string, observeFunc func
 	if err != nil {
 		return nil, fmt.Errorf("cannot create observe request: %w", err)
 	}
+	defer pool.ReleaseMessage(req)
 	token := req.Token()
 	req.SetObserve(0)
 	o := &Observation{
@@ -99,7 +97,19 @@ func (cc *ClientConn) Observe(ctx context.Context, path string, observeFunc func
 	}
 	respCodeChan := make(chan codes.Code, 1)
 	waitForReponse := uint32(1)
-	cc.observationRequests.Store(token.String(), req)
+
+	options, err := req.Options().Clone()
+	if err != nil {
+		return nil, fmt.Errorf("cannot clone options: %w", err)
+	}
+
+	obs := message.Message{
+		Context: req.Context(),
+		Token:   req.Token(),
+		Code:    req.Code(),
+		Options: options,
+	}
+	cc.observationRequests.Store(token.String(), obs)
 	err = o.cc.observationTokenHandler.Insert(token.String(), func(w *ResponseWriter, r *pool.Message) {
 		code := r.Code()
 		if atomic.CompareAndSwapUint32(&waitForReponse, 1, 0) {

tcp/message/pool/message.go

@@ -138,13 +138,7 @@ func ConvertFrom(m *message.Message) (*Message, error) {
 
 // ConvertTo converts pool message to common message.
 func ConvertTo(m *Message) (*message.Message, error) {
-	opts := make(message.Options, 0, len(m.Options()))
-	buf := make([]byte, 64)
-	opts, used, err := opts.ResetOptionsTo(buf, m.Options())
-	if err == message.ErrTooSmall {
-		buf = append(buf, make([]byte, used-len(buf))...)
-		opts, used, err = opts.ResetOptionsTo(buf, m.Options())
-	}
+	opts, err := m.Options().Clone()
 	if err != nil {
 		return nil, err
 	}

udp/message/pool/message.go

@@ -178,13 +178,7 @@ func ConvertFrom(m *message.Message) (*Message, error) {
 
 // ConvertTo converts pool message to common message.
 func ConvertTo(m *Message) (*message.Message, error) {
-	opts := make(message.Options, 0, len(m.Options()))
-	buf := make([]byte, 64)
-	opts, used, err := opts.ResetOptionsTo(buf, m.Options())
-	if err == message.ErrTooSmall {
-		buf = append(buf, make([]byte, used-len(buf))...)
-		opts, used, err = opts.ResetOptionsTo(buf, m.Options())
-	}
+	opts, err := m.Options().Clone()
 	if err != nil {
 		return nil, err
 	}