Skip to content

Commit

Permalink
sync from kubeedge/kubedge repo
Browse files Browse the repository at this point in the history
Signed-off-by: fisherxu <[email protected]>
  • Loading branch information
fisherxu committed May 31, 2021
1 parent cd19bca commit b165d75
Show file tree
Hide file tree
Showing 5 changed files with 64 additions and 52 deletions.
1 change: 1 addition & 0 deletions OWNERS
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,4 @@ reviewers:
- kadisi
- subpathdev
- anyushun
- Iceber
11 changes: 5 additions & 6 deletions pkg/common/util/conn.go
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,12 @@ type UnixSocket struct {
}

// NewUnixSocket create new socket
func NewUnixSocket(filename string, size ...int) *UnixSocket {
size1 := 10480
if size != nil {
size1 = size[0]
func NewUnixSocket(filename string, sizes ...int) *UnixSocket {
size := 10480
if len(sizes) != 0 {
size = sizes[0]
}
us := UnixSocket{filename: filename, bufsize: size1}
return &us
return &UnixSocket{filename: filename, bufsize: size}
}

func (us *UnixSocket) createServer() {
Expand Down
39 changes: 17 additions & 22 deletions pkg/core/context/context_channel.go
Original file line number Diff line number Diff line change
Expand Up @@ -160,22 +160,19 @@ func (ctx *ChannelContext) SendResp(message model.Message) {

// SendToGroup send msg to modules. Todo: do not stuck
func (ctx *ChannelContext) SendToGroup(moduleType string, message model.Message) {
// avoid exception because of channel closing
// TODO: need reconstruction
defer func() {
if exception := recover(); exception != nil {
klog.Warningf("Recover when sendToGroup message, exception: %+v", exception)
}
}()

send := func(ch chan model.Message) {
// avoid exception because of channel closing
// TODO: need reconstruction
defer func() {
if exception := recover(); exception != nil {
klog.Warningf("Recover when sendToGroup message, exception: %+v", exception)
}
}()
select {
case ch <- message:
default:
klog.Warningf("the message channel is full, message: %+v", message)
select {
case ch <- message:
}
ch <- message
}
}
if channelList := ctx.getTypeChannel(moduleType); channelList != nil {
Expand All @@ -190,14 +187,6 @@ func (ctx *ChannelContext) SendToGroup(moduleType string, message model.Message)
// SendToGroupSync : broadcast the message to echo module channel, the module send response back anon channel
// check timeout and the size of anon channel
func (ctx *ChannelContext) SendToGroupSync(moduleType string, message model.Message, timeout time.Duration) error {
// avoid exception because of channel closing
// TODO: need reconstruction
defer func() {
if exception := recover(); exception != nil {
klog.Warningf("Recover when sendToGroupsync message, exception: %+v", exception)
}
}()

if timeout <= 0 {
timeout = MessageTimeoutDefault
}
Expand Down Expand Up @@ -242,6 +231,13 @@ func (ctx *ChannelContext) SendToGroupSync(moduleType string, message model.Mess

var timeoutCounter int32
send := func(ch chan model.Message) {
// avoid exception because of channel closing
// TODO: need reconstruction
defer func() {
if exception := recover(); exception != nil {
klog.Warningf("Recover when sendToGroupsync message, exception: %+v", exception)
}
}()
sendTimer := time.NewTimer(time.Until(deadline))
select {
case ch <- message:
Expand Down Expand Up @@ -308,13 +304,12 @@ func (ctx *ChannelContext) addChannel(module string, moduleCh chan model.Message
func (ctx *ChannelContext) delChannel(module string) {
// delete module channel from channels map
ctx.chsLock.Lock()
_, exist := ctx.channels[module]
if !exist {
if _, exist := ctx.channels[module]; !exist {
ctx.chsLock.Unlock()
klog.Warningf("Failed to get channel, module:%s", module)
return
}
delete(ctx.channels, module)

ctx.chsLock.Unlock()

// delete module channel from typechannels map
Expand Down
23 changes: 11 additions & 12 deletions pkg/core/core.go
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,9 @@ func StartModules() {

modules := GetModules()
for name, module := range modules {
//Init the module
// Init the module
beehiveContext.AddModule(name)
//Assemble typeChannels for sendToGroup
// Assemble typeChannels for sendToGroup
beehiveContext.AddModuleGroup(name, module.Group())
go module.Start()
klog.Infof("Starting module %v", name)
Expand All @@ -30,16 +30,15 @@ func GracefulShutdown() {
c := make(chan os.Signal)
signal.Notify(c, syscall.SIGINT, syscall.SIGHUP, syscall.SIGTERM,
syscall.SIGQUIT, syscall.SIGILL, syscall.SIGTRAP, syscall.SIGABRT)
select {
case s := <-c:
klog.Infof("Get os signal %v", s.String())
//Cleanup each modules
beehiveContext.Cancel()
modules := GetModules()
for name, _ := range modules {
klog.Infof("Cleanup module %v", name)
beehiveContext.Cleanup(name)
}
s := <-c
klog.Infof("Get os signal %v", s.String())

// Cleanup each modules
beehiveContext.Cancel()
modules := GetModules()
for name := range modules {
klog.Infof("Cleanup module %v", name)
beehiveContext.Cleanup(name)
}
}

Expand Down
42 changes: 30 additions & 12 deletions pkg/core/model/message.go
Original file line number Diff line number Diff line change
@@ -1,6 +1,8 @@
package model

import (
"encoding/json"
"fmt"
"time"

uuid "github.com/satori/go.uuid"
Expand All @@ -15,13 +17,16 @@ const (
ResponseOperation = "response"
ResponseErrorOperation = "error"

ResourceTypePod = "pod"
ResourceTypeConfigmap = "configmap"
ResourceTypeSecret = "secret"
ResourceTypeNode = "node"
ResourceTypePodlist = "podlist"
ResourceTypePodStatus = "podstatus"
ResourceTypeNodeStatus = "nodestatus"
ResourceTypePod = "pod"
ResourceTypeConfigmap = "configmap"
ResourceTypeSecret = "secret"
ResourceTypeNode = "node"
ResourceTypePodlist = "podlist"
ResourceTypePodStatus = "podstatus"
ResourceTypeNodeStatus = "nodestatus"
ResourceTypeRule = "rule"
ResourceTypeRuleEndpoint = "ruleendpoint"
ResourceTypeRuleStatus = "rulestatus"
)

// Message struct
Expand Down Expand Up @@ -118,27 +123,40 @@ func (msg *Message) GetID() string {
return msg.Header.ID
}

//GetParentID returns message parent id
// GetParentID returns message parent id
func (msg *Message) GetParentID() string {
return msg.Header.ParentID
}

//GetTimestamp returns message timestamp
// GetTimestamp returns message timestamp
func (msg *Message) GetTimestamp() int64 {
return msg.Header.Timestamp
}

//GetContent returns message content
// GetContent returns message content
func (msg *Message) GetContent() interface{} {
return msg.Content
}

//GetResourceVersion returns message resource version
// GetContentData returns message content data
func (msg *Message) GetContentData() ([]byte, error) {
if data, ok := msg.Content.([]byte); ok {
return data, nil
}

data, err := json.Marshal(msg.Content)
if err != nil {
return nil, fmt.Errorf("marshal message content failed: %s", err)
}
return data, nil
}

// GetResourceVersion returns message resource version
func (msg *Message) GetResourceVersion() string {
return msg.Header.ResourceVersion
}

//UpdateID returns message object updating its ID
// UpdateID returns message object updating its ID
func (msg *Message) UpdateID() *Message {
msg.Header.ID = uuid.NewV4().String()
return msg
Expand Down

0 comments on commit b165d75

Please sign in to comment.