martino/uopi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

working epics ioc and tested

Browse Source
This commit is contained in:
Martino Ferrari
2026-05-04 21:13:36 +02:00
parent 90669c5fd6
commit 0a5a85e4c4
25 changed files with 1550 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/ca/conn.go
+45 -16
View File
@@ -57,13 +57,15 @@ type chanState struct {
cid uint32
pvName string
mu sync.RWMutex
sid uint32 // server-assigned channel ID (0 until CREATE_CHAN reply)
dbfType int // native DBF field type
count uint32 // element count
access uint32 // AccessRead | AccessWrite bitmask
readyC chan struct{} // closed once CREATE_CHAN reply is received
monitors []*monState // active subscriptions
mu sync.RWMutex
sid uint32 // server-assigned channel ID (0 until CREATE_CHAN reply)
dbfType int // native DBF field type
count uint32 // element count
access uint32 // AccessRead | AccessWrite bitmask
gotChan bool // CREATE_CHAN reply received for current connection
gotAccess bool // ACCESS_RIGHTS received for current connection
readyC chan struct{} // closed once both CREATE_CHAN and ACCESS_RIGHTS received; replaced on reconnect
monitors []*monState // active subscriptions
}
func newChanState(cid uint32, pvName string) *chanState {
@@ -82,6 +84,8 @@ func newChanState(cid uint32, pvName string) *chanState {
func (cs *chanState) resetForReconnect() {
cs.mu.Lock()
cs.sid = 0
cs.gotChan = false
cs.gotAccess = false
old := cs.readyC
cs.readyC = make(chan struct{})
cs.mu.Unlock()
@@ -94,25 +98,25 @@ func (cs *chanState) resetForReconnect() {
}
}
// waitReady blocks until sid != 0 (CREATE_CHAN reply received for the current
// connection) or ctx expires. It loops through reconnections automatically:
// when resetForReconnect closes the current readyC the goroutine wakes, sees
// sid == 0, and waits on the freshly created channel.
// waitReady blocks until both CREATE_CHAN and ACCESS_RIGHTS have been received
// (i.e. sid != 0 and gotAccess == true) or ctx expires. It loops through
// reconnections automatically: when resetForReconnect closes the current readyC
// the goroutine wakes, sees sid == 0, and waits on the freshly created channel.
func (cs *chanState) waitReady(ctx context.Context) error {
for {
cs.mu.RLock()
sid := cs.sid
gotAccess := cs.gotAccess
ready := cs.readyC
cs.mu.RUnlock()
if sid != 0 {
if sid != 0 && gotAccess {
return nil
}
select {
case <-ready:
// State changed — either CREATE_CHAN reply (sid set) or reconnect
// started (sid cleared, new readyC installed). Loop to check.
// State changed — loop to re-check sid and gotAccess.
case <-ctx.Done():
return ctx.Err()
}
@@ -378,7 +382,11 @@ func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) {
cs.sid = hdr.Parameter2
cs.dbfType = int(hdr.DataType)
cs.count = hdr.DataCount
ready := cs.readyC
cs.gotChan = true
var ready chan struct{}
if cs.gotAccess {
ready = cs.readyC
}
// Snapshot monitors for EVENT_ADD re-subscription.
mons := make([]*monState, len(cs.monitors))
copy(mons, cs.monitors)
@@ -397,7 +405,12 @@ func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) {
default:
}
}
close(ready) // unblock waitReady callers
// Unblock waitReady only once ACCESS_RIGHTS has also arrived.
// ACCESS_RIGHTS always follows CREATE_CHAN, so ready is typically nil here
// and the close happens in the CmdAccessRights handler below.
if ready != nil {
close(ready)
}
case proto.CmdCreateFail:
// Server does not host this PV (or quota exceeded).
@@ -422,13 +435,29 @@ func (c *circuit) dispatch(conn net.Conn, hdr proto.Header, payload []byte) {
case proto.CmdAccessRights:
// Parameter1 = cid, Parameter2 = access bitmask.
// ACCESS_RIGHTS always arrives just after CREATE_CHAN; we defer closing
// readyC until here so callers see the correct access value immediately.
c.mu.RLock()
cs, ok := c.byCID[hdr.Parameter1]
c.mu.RUnlock()
dbg("CA ACCESS_RIGHTS", "cid", hdr.Parameter1, "access", hdr.Parameter2, "found", ok)
if ok {
cs.mu.Lock()
cs.access = hdr.Parameter2
cs.gotAccess = true
var ready chan struct{}
if cs.gotChan {
ready = cs.readyC
}
cs.mu.Unlock()
if ready != nil {
select {
case <-ready:
// Already closed (guard against duplicate ACCESS_RIGHTS).
default:
close(ready)
}
}
}
case proto.CmdEventAdd: