tried simplified protocol, testing

This commit is contained in:
MahnoKropotkinvich 2024-12-25 18:53:10 +08:00
parent 426adf8d3a
commit cbce5f7c95
3 changed files with 31 additions and 40 deletions

10
conn.go
View File

@ -33,7 +33,7 @@ func bridge() {
clientSignalChan.TX <- msg clientSignalChan.TX <- msg
} }
} }
func handle(clientOriginal net.Conn) { func handleConn(clientOriginal net.Conn) {
client := timeoutConn{clientOriginal} client := timeoutConn{clientOriginal}
defer clientOriginal.Close() defer clientOriginal.Close()
queryChan := make(QueryChan) queryChan := make(QueryChan)
@ -66,18 +66,16 @@ func handle(clientOriginal net.Conn) {
wg.Wait() wg.Wait()
GetLogger().Infof("Connection from %s closed", clientOriginal.RemoteAddr()) GetLogger().Infof("Connection from %s closed", clientOriginal.RemoteAddr())
} }
CntChan <- INCREASE
defer func() { CntChan <- DECREASE }()
queryChan <- STOPPED queryChan <- STOPPED
st, _ := <-queryChan st, _ := <-queryChan
switch st { switch st {
case RUNNING: case RUNNING:
CntChan <- INCREASE
defer func() { CntChan <- DECREASE }()
proceed() proceed()
case STOPPED, WAITING: case STOPPED, WAITING:
// client.Write([]byte("Server not ready!")) // client.Write([]byte("Server not ready!"))
GetLogger().Infof("Connection queued, server currently at %s state", stateToStr[state]) GetLogger().Infof("Connection queued, server currently at %s state", stateToStr[state])
CntChan <- INCREASE
defer func() { CntChan <- DECREASE }()
curChan := make(chan MCState) curChan := make(chan MCState)
defer close(curChan) defer close(curChan)
clientSignalChan.Add(curChan) clientSignalChan.Add(curChan)
@ -106,6 +104,6 @@ func Listen() {
} }
//conn.SetReadDeadline(time.Now().Add(time.Duration(config.ConnectTimeout) * time.Second)) //conn.SetReadDeadline(time.Now().Add(time.Duration(config.ConnectTimeout) * time.Second))
GetLogger().Infof("New connection from %s", conn.RemoteAddr()) GetLogger().Infof("New connection from %s", conn.RemoteAddr())
go handle(conn) go handleConn(conn)
} }
} }

57
ds.go
View File

@ -79,26 +79,22 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
modify: make(chan modify_t[T]), modify: make(chan modify_t[T]),
} }
mode := m mode := m
reloadTX := make(chan struct{}) reload := make(chan struct{})
reloadRX := make(chan struct{})
used := make(map[int]bool) used := make(map[int]bool)
unused := make(map[int]bool) unused := make(map[int]bool)
delta := -1 // delta := -1
list := make([]chan T, 0) list := make([]chan T, 0)
selectCases := make([]reflect.SelectCase, 1) selectCases := make([]reflect.SelectCase, 1)
viewCnt := 0
addChan := make(chan int, 1)
go func() { go func() {
for { for {
modification, _ := <-ret.modify modification, _ := <-ret.modify
op := modification.op op := modification.op
id := modification.id id := modification.id
ch := modification.ch ch := modification.ch
reloadTX <- struct{}{} for range viewCnt {
if mode == 2 { reload <- struct{}{}
reloadTX <- struct{}{}
}
<-reloadRX // ok it's safe to do operation
if mode == 2 {
<-reloadRX
} }
switch op { switch op {
case ADD: case ADD:
@ -108,7 +104,7 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
Send: reflect.Value{}, Send: reflect.Value{},
} }
if len(unused) == 0 { if len(unused) == 0 {
delta = len(list) addChan <- len(list)
used[len(list)] = true used[len(list)] = true
list = append(list, ch) list = append(list, ch)
selectCases = append(selectCases, nelem) selectCases = append(selectCases, nelem)
@ -118,7 +114,7 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
which = x which = x
break break
} }
delta = which addChan <- which
used[which] = true used[which] = true
delete(unused, which) delete(unused, which)
list[which] = ch list[which] = ch
@ -135,21 +131,17 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
} }
//deleted <- struct{}{} // <-reloadTX //extra communication //deleted <- struct{}{} // <-reloadTX //extra communication
} }
reloadTX <- struct{}{} for range viewCnt {
if mode == 2 { reload <- struct{}{}
reloadTX <- struct{}{} //I'm finished
} }
<-reloadRX
if mode == 2 {
<-reloadRX //ok I'll cleanup
}
delta = -1
} }
}() }()
go func() { go func() {
viewCnt++
selectCases[0] = reflect.SelectCase{ selectCases[0] = reflect.SelectCase{
Dir: reflect.SelectRecv, Dir: reflect.SelectRecv,
Chan: reflect.ValueOf(reloadTX), Chan: reflect.ValueOf(reload),
} }
prevId := -1 prevId := -1
for { for {
@ -164,10 +156,10 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
// //
// delete(used, id) // delete(used, id)
ret.modify <- modify_t[T]{DELETE, id - 1, nil} ret.modify <- modify_t[T]{DELETE, id - 1, nil}
<-reloadTX <-reload
reloadRX <- struct{}{} //I'm currently not dealing with other chan // reloadRX <- struct{}{} //I'm currently not dealing with other chan
<-reloadTX //waiting for you finished <-reload //waiting for you finished
reloadRX <- struct{}{} //I've read all deltas, you can release them // reloadRX <- struct{}{} //I've read all deltas, you can release them
continue continue
} }
if mode == 1 && id != 0 { if mode == 1 && id != 0 {
@ -179,9 +171,9 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
prevId = id prevId = id
} }
} else if id == 0 { } else if id == 0 {
reloadRX <- struct{}{} //I'm currently not dealing with other chan // reloadRX <- struct{}{} //I'm currently not dealing with other chan
<-reloadTX //waiting for you finished <-reload //waiting for you finished
reloadRX <- struct{}{} //I've read all deltas, you can release them // reloadRX <- struct{}{} //I've read all deltas, you can release them
} }
} }
}() }()
@ -190,15 +182,16 @@ func NewDynamicMultiChan[T constraints.Ordered](reply bool, m int) *DynamicMulti
msgList := make([]T, 0) msgList := make([]T, 0)
for { for {
select { select {
case <-reloadTX: case <-reload:
reloadRX <- struct{}{} //I'm currently not dealing with other chan // reloadRX <- struct{}{} //I'm currently not dealing with other chan
<-reloadTX //waiting for you finished <-reload //waiting for you finished
delta, _ := <-addChan
if delta != -1 { if delta != -1 {
for _, x := range msgList { for _, x := range msgList {
list[delta] <- x list[delta] <- x
} }
} }
reloadRX <- struct{}{} //I've read all deltas, you can release them // reloadRX <- struct{}{} //I've read all deltas, you can release them
case msg, _ := <-ret.TX: case msg, _ := <-ret.TX:
for id, ch := range list { for id, ch := range list {
if used[id] { if used[id] {

4
fsm.go
View File

@ -137,7 +137,7 @@ func handleState() { //goroutine only, handles both write and read
handleWaitingToRunning() handleWaitingToRunning()
//case RUNNING: //case RUNNING:
default: default:
panic("handleState():written wrong!" + stateToStr[state]) //panic("handleState():written wrong!" + stateToStr[state])
} }
case EMPTY: case EMPTY:
switch state { switch state {
@ -145,7 +145,7 @@ func handleState() { //goroutine only, handles both write and read
handleRunningToWaiting() handleRunningToWaiting()
//case BOOTING: //case BOOTING:
default: default:
panic("handleState():written wrong!" + stateToStr[state]) //panic("handleState():written wrong!" + stateToStr[state])
} }
} }
case <-multiChan.RX: case <-multiChan.RX: