1, 优化sshd的客户端代码, 增强自身的鲁棒性(命令发出后，就算服务端没有应答，也不会一直无限期死等), 客户端版本升级到1.0.0.2；2, 优化设置系统时间的函数

netmgrd/ntp.go

@@ -5,6 +5,7 @@ import (
 	"math"
 	"os/exec"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/beevik/ntp"
@@ -91,10 +92,16 @@ func SetSystemTime(t time.Time) error {
 	setTimeMu.Lock()
 	defer setTimeMu.Unlock()
 
-	cmd := exec.Command("date", "-s", t.Local().Format("2006-01-02 15:04:05"))
-	output, err := cmd.CombinedOutput()
-	if err != nil {
-		return fmt.Errorf("%v, output: %s", err, string(output))
+	tv := syscall.Timeval{
+		Sec:  t.Unix(),
+		Usec: int64(t.Nanosecond() / 1000),
+	}
+
+	if err := syscall.Settimeofday(&tv); err != nil {
+		return err
 	}
+
+	_ = exec.Command("hwclock", "-w").Run()
+
 	return nil
 }

sshd/client/client.go

@@ -25,7 +25,7 @@ const MODULE_NAME = "YFKJ_SSH_CLIENT"
 
 var (
 	coupler               *MQTTCoupler
-	Version               = "1.0.0.1"
+	Version               = "1.0.0.2"
 	BuildTime             = "unknown"
 	ErrBrokerAddressEmpty = errors.New("mqtt server address is empty")
 	ErrIMEINotAvailable   = errors.New("device imei is not available")

sshd/client/coupler.go

@@ -11,6 +11,7 @@ import (
 
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"hnyfkj.com.cn/rtu/linux/utils/jsonrpc2"
+	"hnyfkj.com.cn/rtu/linux/utils/shell"
 )
 
 const (
@@ -19,6 +20,13 @@ const (
 	SlowInterval      = 5 * time.Second //// 慢速检测时间间隔
 )
 
+type pendingRequest struct {
+	ch   chan *jsonrpc2.Response
+	done chan struct{} // 命令请求完成信号: 已应答/超时/被中断
+	// 命令请求超时/中断后, 使得等待应答的携程能收到结束信号退出
+	// !避免请求端已经放弃等待, 但应答端仍然在等待结果并写入通道
+}
+
 type MQTTCoupler struct {
 	ctx    context.Context
 	cancel context.CancelFunc
@@ -36,9 +44,9 @@ type MQTTCoupler struct {
 	cwd      string // 当前工作目录
 
 	// 交互式命令高频执行, 为了性能上的考虑-这里不使用"sync.Map"
-	cmdMu     sync.Mutex                        // 串行执行锁
-	pending   map[int64]chan *jsonrpc2.Response // 等命令结果
-	pendingMu sync.Mutex                        // 等待结果锁
+	cmdMu     sync.Mutex                // 串行执行的锁LLLLLL
+	pending   map[int64]*pendingRequest // 等命令的结果RRRRRR
+	pendingMu sync.Mutex                // 等待结果的锁LLLLLL
 
 	interrupted chan struct{} // Ctrl+C 通知当前命令取消的信号
 }
@@ -78,7 +86,7 @@ func (c *MQTTCoupler) init2() error {
 		}
 	}
 
-	c.pending = make(map[int64]chan *jsonrpc2.Response)
+	c.pending = make(map[int64]*pendingRequest)
 
 	c.client = mqtt.NewClient(opts)
 	go c.keepOnline()
@@ -146,15 +154,19 @@ func (c *MQTTCoupler) doCmd(method string, params any, id ...int64) (*jsonrpc2.R
 		return zero, err
 	}
 
-	ch := make(chan *jsonrpc2.Response, 1)
+	pr := &pendingRequest{
+		ch:   make(chan *jsonrpc2.Response, 1),
+		done: make(chan struct{}),
+	}
 
 	c.pendingMu.Lock()
-	c.pending[reqID] = ch
+	c.pending[reqID] = pr
 	c.pendingMu.Unlock()
 	defer func() {
 		c.pendingMu.Lock()
 		delete(c.pending, reqID)
 		c.pendingMu.Unlock()
+		close(pr.done)
 	}()
 
 	token := c.client.Publish(c.pubTopic, MqttQos1, false, b)
@@ -170,26 +182,32 @@ func (c *MQTTCoupler) doCmd(method string, params any, id ...int64) (*jsonrpc2.R
 	}
 
 	if c.isCtrlCommand(method) { // 控制指令, 等待结果或超时
-		var timer *time.Timer
-		var timeout <-chan time.Time
-		timer = time.NewTimer(3 * time.Second)
-		timeout = timer.C
+		timer := time.NewTimer(time.Duration(3) * time.Second)
 		defer timer.Stop()
 
 		select {
 		case <-c.ctx.Done():
 			return zero, c.ctx.Err()
-		case resp := <-ch:
+		case resp := <-pr.ch:
 			return resp, nil
-		case <-timeout:
+		case <-timer.C:
 			return zero, fmt.Errorf("command timeout")
 		}
 	} else { // 用户指令, 等待结果或用户主动中断取消: Ctrl+C
+		v, ok := c.getCmdTimeout(params)
+		if !ok || v <= 0 { // 如果没有指定超时, 则使用默认超时
+			v = int(shell.DefaultTimeout/time.Second) + 1
+		}
+		timer := time.NewTimer(time.Duration(v+1) * time.Second)
+		defer timer.Stop()
+
 		select {
 		case <-c.ctx.Done():
 			return zero, c.ctx.Err()
-		case resp := <-ch:
+		case resp := <-pr.ch:
 			return resp, nil
+		case <-timer.C:
+			return zero, fmt.Errorf("command timeout")
 		case <-c.interrupted:
 			return zero, fmt.Errorf("command interrupted by user")
 		}
@@ -210,7 +228,7 @@ func (c *MQTTCoupler) onCmdAck(client mqtt.Client, msg mqtt.Message) {
 	respID := *resp.ID
 
 	c.pendingMu.Lock()
-	ch, ok := c.pending[respID]
+	pr, ok := c.pending[respID]
 	c.pendingMu.Unlock()
 
 	if !ok { /////////////// 未找到对应的请求, 忽略不管
@@ -218,7 +236,7 @@ func (c *MQTTCoupler) onCmdAck(client mqtt.Client, msg mqtt.Message) {
 	}
 
 	select {
-	case ch <- resp:
-	default:
+	case pr.ch <- resp:
+	case <-pr.done:
 	}
 }

sshd/client/invoker.go

@@ -50,13 +50,17 @@ func (c *MQTTCoupler) ping() (*jsonrpc2.Response, error) {
 	return c.doCmd(rpc_ping, params)
 }
 
+// 用户请求命令
+type ShellExecRequest struct {
+	ClientID string `json:"client_id"`
+	/// 命令执行的参数, 以匿名的方式嵌套
+	shell.ExecuteParams
+}
+
 // 执行一条命令
 func (c *MQTTCoupler) exec(
 	cmd string) (*shell.ExecuteResult, error) {
-	params := struct {
-		ClientID string `json:"client_id"`
-		shell.ExecuteParams
-	}{
+	params := ShellExecRequest{
 		ClientID: c.clientID,
 		ExecuteParams: shell.ExecuteParams{
 			HostFingerprint: c.machineID, // 调用者主机指纹
@@ -88,6 +92,20 @@ func (c *MQTTCoupler) exec(
 	return &exrs, nil
 }
 
+// 取出命令超时
+func (c *MQTTCoupler) getCmdTimeout(params any) (int, bool) {
+	switch v := params.(type) {
+	case ShellExecRequest:
+		return v.Timeout, true
+	case *ShellExecRequest:
+		if v == nil {
+			return 0, false
+		}
+		return v.Timeout, true
+	}
+	return 0, false
+}
+
 // 中断当前执行
 func (c *MQTTCoupler) stop() (*jsonrpc2.Response, error) {
 	params := struct {

utils/ftpclient/cmd/main.go

@@ -1 +0,0 @@
-package main