Repository: lf-edge/ekuiper
Pull request #1205 (2022-03-11 10:54:54)

Portable runtime timeout handling


Patch:
diff --git a/deploy/docker/Dockerfile-slim b/deploy/docker/Dockerfile-slim
index 10624e84..ecdd1769 100644
--- a/deploy/docker/Dockerfile-slim
+++ b/deploy/docker/Dockerfile-slim
@@ -21,7 +21,7 @@ WORKDIR /go/kuiper
 RUN apt update && apt install -y pkg-config libczmq-dev
 RUN make build_with_edgex
 
-FROM debian:11
+FROM debian:buster-20211011 
 
 COPY ./deploy/docker/docker-entrypoint.sh /usr/bin/docker-entrypoint.sh
 COPY --from=builder /go/kuiper/_build/kuiper-* /kuiper/
diff --git a/internal/plugin/portable/runtime/connection.go b/internal/plugin/portable/runtime/connection.go
index b8f541f3..22f6baa0 100644
--- a/internal/plugin/portable/runtime/connection.go
+++ b/internal/plugin/portable/runtime/connection.go
@@ -31,10 +31,6 @@ import (
 
 // sockOptions Initialized in config
 var (
-	sockOptions = map[string]interface{}{
-		mangos.OptionRecvDeadline: 1 * time.Hour,
-		mangos.OptionRetryTime:    0,
-	}
 	dialOptions = map[string]interface{}{
 		mangos.OptionDialAsynch:       false,
 		mangos.OptionMaxReconnectTime: 5 * time.Second,
@@ -147,7 +143,9 @@ func CreateSourceChannel(ctx api.StreamContext) (DataInChannel, error) {
 	if sock, err = pull.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new pull socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRecvDeadline: 500 * time.Millisecond,
+	})
 	url := fmt.Sprintf("ipc:///tmp/%s_%s_%d.ipc", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	if err = listenWithRetry(sock, url); err != nil {
 		return nil, fmt.Errorf("can't listen on pull socket for %s: %s", url, err.Error())
@@ -164,7 +162,12 @@ func CreateFunctionChannel(symbolName string) (DataReqChannel, error) {
 	if sock, err = rep.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new rep socket: %s", err)
 	}
-	setSockOptions(sock)
+	// Function must send out data quickly and wait for the response with some buffer
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRecvDeadline: 5000 * time.Millisecond,
+		mangos.OptionSendDeadline: 1000 * time.Millisecond,
+		mangos.OptionRetryTime:    0,
+	})
 	url := fmt.Sprintf("ipc:///tmp/func_%s.ipc", symbolName)
 	if err = listenWithRetry(sock, url); err != nil {
 		return nil, fmt.Errorf("can't listen on rep socket for %s: %s", url, err.Error())
@@ -181,7 +184,9 @@ func CreateSinkChannel(ctx api.StreamContext) (DataOutChannel, error) {
 	if sock, err = push.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new push socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionSendDeadline: 1000 * time.Millisecond,
+	})
 	url := fmt.Sprintf("ipc:///tmp/%s_%s_%d.ipc", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	if err = sock.DialOptions(url, dialOptions); err != nil {
 		return nil, fmt.Errorf("can't dial on push socket: %s", err.Error())
@@ -198,7 +203,12 @@ func CreateControlChannel(pluginName string) (ControlChannel, error) {
 	if sock, err = rep.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new rep socket: %s", err)
 	}
-	setSockOptions(sock)
+	// NO time out now for control channel
+	// because the plugin instance liveness can be detected
+	// thus, if the plugin exit, the control channel will be closed
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRecvDeadline: 1 * time.Hour,
+	})
 	url := fmt.Sprintf("ipc:///tmp/plugin_%s.ipc", pluginName)
 	if err = listenWithRetry(sock, url); err != nil {
 		return nil, fmt.Errorf("can't listen on rep socket: %s", err.Error())
@@ -207,10 +217,10 @@ func CreateControlChannel(pluginName string) (ControlChannel, error) {
 	return &NanomsgReqChannel{sock: sock}, nil
 }
 
-func setSockOptions(sock mangos.Socket) {
+func setSockOptions(sock mangos.Socket, sockOptions map[string]interface{}) {
 	for k, v := range sockOptions {
 		err := sock.SetOption(k, v)
-		if err != nil {
+		if err != nil && err != mangos.ErrBadOption {
 			conf.Log.Errorf("can't set socket option %s: %s", k, err.Error())
 		}
 	}
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager.go b/internal/plugin/portable/runtime/plugin_ins_manager.go
index c76ed5ad..2488e169 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager.go
@@ -109,7 +109,7 @@ func (i *PluginIns) Stop() error {
 	if i.ctrlChan != nil {
 		err = i.ctrlChan.Close()
 	}
-	if i.process != nil {
+	if i.process != nil { // will also trigger process exit clean up
 		err = i.process.Kill()
 	}
 	return err
@@ -150,6 +150,10 @@ func (p *pluginInsManager) AddPluginIns(name string, ins *PluginIns) {
 	p.instances[name] = ins
 }
 
+// getOrStartProcess Control the plugin process lifecycle.
+// Need to manage the resources: instances map, control socket, plugin process
+// 1. During creation, clean up those resources for any errors in defer immediately after the resource is created.
+// 2. During plugin running, when detecting plugin process exit, clean up those resources for the current ins.
 func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *PortableConfig) (*PluginIns, error) {
 	p.Lock()
 	defer p.Unlock()
@@ -204,9 +208,11 @@ func (p *pluginInsManager) getOrStartProcess(pluginMeta *PluginMeta, pconf *Port
 		if err != nil {
 			conf.Log.Printf("plugin executable %s stops with error %v", pluginMeta.Executable, err)
 		}
-
-		if ins, ok := p.getPluginIns(pluginMeta.Name); ok {
-			_ = ins.ctrlChan.Close()
+		// must make sure the plugin ins is not cleaned up yet by checking the process identity
+		if ins, ok := p.getPluginIns(pluginMeta.Name); ok && ins.process == cmd.Process {
+			if ins.ctrlChan != nil {
+				_ = ins.ctrlChan.Close()
+			}
 			p.deletePluginIns(pluginMeta.Name)
 		}
 	}()
diff --git a/internal/plugin/portable/runtime/source.go b/internal/plugin/portable/runtime/source.go
index 44c382d2..8df3eec7 100644
--- a/internal/plugin/portable/runtime/source.go
+++ b/internal/plugin/portable/runtime/source.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -99,7 +99,11 @@ func (ps *PortableSource) Open(ctx api.StreamContext, consumer chan<- api.Source
 			return
 		case mangos.ErrRecvTimeout:
 			ctx.GetLogger().Debug("source receive timeout, retry")
-			continue
+			select {
+			case <-ctx.Done():
+				ctx.GetLogger().Info("stop source")
+				return
+			}
 		case nil:
 			// do nothing
 		default:
diff --git a/sdk/go/connection/connection.go b/sdk/go/connection/connection.go
index ffbd37d4..7058390e 100644
--- a/sdk/go/connection/connection.go
+++ b/sdk/go/connection/connection.go
@@ -28,9 +28,6 @@ import (
 
 // Options Initialized in plugin.go Start according to the config
 var (
-	SockOptions = map[string]interface{}{
-		mangos.OptionRetryTime: 0,
-	}
 	dialOptions = map[string]interface{}{
 		mangos.OptionDialAsynch:       false,
 		mangos.OptionMaxReconnectTime: 5 * time.Second,
@@ -100,7 +97,9 @@ func CreateControlChannel(pluginName string) (ControlChannel, error) {
 	if sock, err = req.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new req socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRetryTime: 0,
+	})
 	url := fmt.Sprintf("ipc:///tmp/plugin_%s.ipc", pluginName)
 	if err = sock.DialOptions(url, dialOptions); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())
@@ -116,7 +115,9 @@ func CreateSourceChannel(ctx api.StreamContext) (DataOutChannel, error) {
 	if sock, err = push.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new push socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionSendDeadline: 1000 * time.Millisecond,
+	})
 	url := fmt.Sprintf("ipc:///tmp/%s_%s_%d.ipc", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	if err = sock.DialOptions(url, dialOptions); err != nil {
 		return nil, fmt.Errorf("can't dial on push socket: %s", err.Error())
@@ -132,7 +133,11 @@ func CreateFuncChannel(symbolName string) (DataInOutChannel, error) {
 	if sock, err = req.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new req socket: %s", err)
 	}
-	setSockOptions(sock)
+	// The recv should not have timeout because it is event driven
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionSendDeadline: 1000 * time.Millisecond,
+		mangos.OptionRetryTime:    0,
+	})
 	url := fmt.Sprintf("ipc:///tmp/func_%s.ipc", symbolName)
 	if err = sock.DialOptions(url, dialOptions); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())
@@ -148,7 +153,9 @@ func CreateSinkChannel(ctx api.StreamContext) (DataInChannel, error) {
 	if sock, err = pull.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new pull socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRecvDeadline: 500 * time.Millisecond,
+	})
 	url := fmt.Sprintf("ipc:///tmp/%s_%s_%d.ipc", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	if err = listenWithRetry(sock, url); err != nil {
 		return nil, fmt.Errorf("can't listen on pull socket for %s: %s", url, err.Error())
@@ -156,11 +163,11 @@ func CreateSinkChannel(ctx api.StreamContext) (DataInChannel, error) {
 	return sock, nil
 }
 
-func setSockOptions(sock mangos.Socket) {
-	for k, v := range SockOptions {
+func setSockOptions(sock mangos.Socket, sockOptions map[string]interface{}) {
+	for k, v := range sockOptions {
 		err := sock.SetOption(k, v)
-		if err != nil {
-			context.Log.Warnf("can't set option %s: %s", k, err.Error())
+		if err != nil && err != mangos.ErrBadOption {
+			context.Log.Errorf("can't set socket option %s: %s", k, err.Error())
 		}
 	}
 }
diff --git a/sdk/go/runtime/plugin.go b/sdk/go/runtime/plugin.go
index 2ee8fe47..e65747f7 100644
--- a/sdk/go/runtime/plugin.go
+++ b/sdk/go/runtime/plugin.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -22,7 +22,6 @@ import (
 	"github.com/lf-edge/ekuiper/sdk/go/api"
 	"github.com/lf-edge/ekuiper/sdk/go/connection"
 	"github.com/lf-edge/ekuiper/sdk/go/context"
-	"go.nanomsg.org/mangos/v3"
 	"os"
 	"os/signal"
 	"sync"
@@ -47,7 +46,6 @@ func initVars(args []string, conf *PluginConfig) {
 		if err != nil {
 			panic(fmt.Sprintf("fail to parse args %v", args))
 		}
-		connection.SockOptions[mangos.OptionSendDeadline] = pc.SendTimeout
 		logger.Infof("config parsed to %v", pc)
 	}
 }
diff --git a/sdk/go/runtime/sink.go b/sdk/go/runtime/sink.go
index 593ad98b..dca9ecc8 100644
--- a/sdk/go/runtime/sink.go
+++ b/sdk/go/runtime/sink.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -63,6 +63,7 @@ func (s *sinkRuntime) run() {
 	for {
 		var msg []byte
 		// blocking read, must be interrupted when stopping
+		// Will be stopped by closing the socket in stop
 		msg, err = s.ch.Recv()
 		switch err {
 		case mangos.ErrClosed:
diff --git a/sdk/python/ekuiper/runtime/connection.py b/sdk/python/ekuiper/runtime/connection.py
index 3cf5c8a2..8bfe5f0f 100644
--- a/sdk/python/ekuiper/runtime/connection.py
+++ b/sdk/python/ekuiper/runtime/connection.py
@@ -54,7 +54,7 @@ class PairChannel:
 class SourceChannel:
 
     def __init__(self, meta: dict):
-        s = Push0()
+        s = Push0(send_timeout=1000)
         url = "ipc:///tmp/{}_{}_{}.ipc".format(meta['ruleId'], meta['opId'], meta['instanceId'])
         logging.info(url)
         dial_with_retry(s, url)


Test patch:
diff --git a/internal/plugin/portable/runtime/connection_test.go b/internal/plugin/portable/runtime/connection_test.go
index 6561e54c..57df589a 100644
--- a/internal/plugin/portable/runtime/connection_test.go
+++ b/internal/plugin/portable/runtime/connection_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -26,6 +26,7 @@ import (
 	"go.nanomsg.org/mangos/v3/protocol/req"
 	"reflect"
 	"testing"
+	"time"
 )
 
 var okMsg = []byte("ok")
@@ -269,7 +270,6 @@ func createMockControlChannel(pluginName string) (*mockControlClient, error) {
 	if sock, err = req.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new req socket: %s", err)
 	}
-	setSockOptions(sock)
 	url := fmt.Sprintf("ipc:///tmp/plugin_%s.ipc", pluginName)
 	if err = sock.Dial(url); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())
@@ -285,7 +285,9 @@ func createMockSourceChannel(ctx api.StreamContext) (mangos.Socket, error) {
 	if sock, err = push.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new push socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRecvDeadline: 500 * time.Millisecond,
+	})
 	url := fmt.Sprintf("ipc:///tmp/%s_%s_%d.ipc", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	if err = sock.Dial(url); err != nil {
 		return nil, fmt.Errorf("can't dial on push socket: %s", err.Error())
@@ -301,7 +303,9 @@ func createMockSinkChannel(ctx api.StreamContext) (mangos.Socket, error) {
 	if sock, err = pull.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new pull socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionSendDeadline: 1000 * time.Millisecond,
+	})
 	url := fmt.Sprintf("ipc:///tmp/%s_%s_%d.ipc", ctx.GetRuleId(), ctx.GetOpId(), ctx.GetInstanceId())
 	if err = listenWithRetry(sock, url); err != nil {
 		return nil, fmt.Errorf("can't listen on pull socket for %s: %s", url, err.Error())
diff --git a/internal/plugin/portable/runtime/plugin_ins_manager_test.go b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
index dfd2de0e..b1250b5a 100644
--- a/internal/plugin/portable/runtime/plugin_ins_manager_test.go
+++ b/internal/plugin/portable/runtime/plugin_ins_manager_test.go
@@ -1,4 +1,4 @@
-// Copyright 2021 EMQ Technologies Co., Ltd.
+// Copyright 2021-2022 EMQ Technologies Co., Ltd.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -177,7 +177,9 @@ func createMockClient(pluginName string) (mangos.Socket, error) {
 	if sock, err = req.NewSocket(); err != nil {
 		return nil, fmt.Errorf("can't get new req socket: %s", err)
 	}
-	setSockOptions(sock)
+	setSockOptions(sock, map[string]interface{}{
+		mangos.OptionRetryTime: 0,
+	})
 	url := fmt.Sprintf("ipc:///tmp/plugin_%s.ipc", pluginName)
 	if err = sock.Dial(url); err != nil {
 		return nil, fmt.Errorf("can't dial on req socket: %s", err.Error())