Implemented epics read/write

internal/datasource/epics/ca_callback.go

@@ -3,6 +3,7 @@
 package epics
 
 /*
+#cgo CFLAGS: -Wno-unused-function
 #include "ca_wrapper.h"
 */
 import "C"

internal/datasource/epics/ca_wrapper.h

@@ -13,7 +13,7 @@
  * goCAConnectionCallback is called when a channel connects or disconnects.
  */
 extern void goCAMonitorCallback(uintptr_t handle, int dbrType, long count,
-                                const void *dbr, int severity,
+                                void *dbr, int severity,
                                 double epicsTimeSecs);
 extern void goCAConnectionCallback(uintptr_t handle, int connected);
 
@@ -78,7 +78,7 @@ static void caMonitorCallbackShim(struct event_handler_args args) {
     }
 
     goCAMonitorCallback((uintptr_t)args.usr, (int)args.type, (long)args.count,
-                        args.dbr, severity, timeSecs);
+                        (void *)args.dbr, severity, timeSecs);
 }
 
 /*
@@ -115,4 +115,29 @@ static int caAddMonitor(chid ch, short dbrType, uintptr_t userHandle,
                         (void *)userHandle, pEvid);
 }
 
+/* Wrappers for ca_get / ca_put, which are macros in cadef.h that expand to
+ * ca_array_get / ca_array_put with count=1.  CGo cannot call macros directly,
+ * so we provide thin static-inline shims here. */
+
+static int caGet(chtype type, chid chan, void *pValue) {
+    return ca_array_get(type, 1u, chan, pValue);
+}
+
+static int caPut(chtype type, chid chan, const void *pValue) {
+    return ca_array_put(type, 1u, chan, pValue);
+}
+
+/* Retrieve the current thread's CA context as an opaque pointer.
+ * Store the result and pass it to caAttachContext() from other threads. */
+static void *caCurrentContext(void) {
+    return (void *)ca_current_context();
+}
+
+/* Attach an existing CA context to the calling thread.  Must be called at the
+ * start of every goroutine (OS thread) that makes CA calls, if that thread did
+ * not call ca_context_create() itself.  Returns ECA_NORMAL on success. */
+static int caAttachContext(void *ctx) {
+    return ca_attach_context((struct ca_client_context *)ctx);
+}
+
 #endif /* CA_WRAPPER_H */

internal/datasource/epics/epics.go

@@ -17,7 +17,7 @@
 package epics
 
 /*
-#cgo CFLAGS: -Wall
+#cgo CFLAGS: -Wall -Wno-unused-function
 #include "ca_wrapper.h"
 #include <stdlib.h>
 */
@@ -26,6 +26,7 @@ import "C"
 import (
 	"context"
 	"fmt"
+	"os"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -68,6 +69,12 @@ type EPICS struct {
 	caAddrList string
 	archiveURL string
 
+	// caCtx is the CA context created in Connect().  Stored as unsafe.Pointer
+	// because the C type (ca_client_context *) is opaque.  Every goroutine
+	// that calls CA functions must call caAttachContext(caCtx) first, because
+	// Go goroutines can run on any OS thread and CA contexts are thread-local.
+	caCtx unsafe.Pointer
+
 	mu       sync.Mutex
 	channels map[string]*caChannel // PV name → CA channel
 	metadata map[string]datasource.Metadata
@@ -98,27 +105,43 @@ func (e *EPICS) Name() string { return "epics" }
 // callbacks are delivered on CA's background threads without needing a
 // ca_pend_event polling loop.
 func (e *EPICS) Connect(_ context.Context) error {
-	// Optionally override EPICS_CA_ADDR_LIST at runtime.
+	// Apply CA address list overrides before creating the context so the
+	// library picks them up during initialisation.
+	// os.Setenv is used instead of C.putenv: putenv takes ownership of the
+	// string pointer, which would race with Go's garbage collector.
 	if e.caAddrList != "" {
-		cs := C.CString(e.caAddrList)
-		defer C.free(unsafe.Pointer(cs))
-		// ca_setenv is not available in all EPICS versions; use putenv via C.
-		envStr := C.CString("EPICS_CA_ADDR_LIST=" + e.caAddrList)
-		defer C.free(unsafe.Pointer(envStr))
-		C.putenv(envStr)
+		os.Setenv("EPICS_CA_ADDR_LIST", e.caAddrList)
+		os.Setenv("EPICS_CA_AUTO_ADDR_LIST", "NO")
 	}
 
 	status := C.ca_context_create(C.ca_enable_preemptive_callback)
 	if status != C.ECA_NORMAL {
 		return fmt.Errorf("epics: ca_context_create failed: status %d", int(status))
 	}
+	// Save the context so goroutines on other OS threads can attach to it.
+	e.caCtx = C.caCurrentContext()
 	return nil
 }
 
-// Subscribe connects to the named PV (if not already connected), installs a
-// monitor, and streams value updates into ch.  The returned CancelFunc removes
-// the monitor and clears the channel.
+// attachCAContext attaches the saved CA context to the calling OS thread.
+// Must be called at the top of every goroutine that uses CA functions,
+// because Go goroutines can be scheduled onto any OS thread.
+// Safe to call redundantly (ECA_ISATTACHED is not an error).
+func (e *EPICS) attachCAContext() {
+	if e.caCtx != nil {
+		C.caAttachContext(e.caCtx)
+	}
+}
+
+// Subscribe creates a CA channel for signal and returns immediately.
+// Connection and monitor setup happen in a background goroutine so that
+// the caller is never blocked waiting for the IOC to respond.
+// A QualityBad value is sent to ch if the channel cannot connect within
+// 60 seconds. The returned CancelFunc tears down the subscription.
 func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasource.Value) (datasource.CancelFunc, error) {
+	// Attach the CA context to this OS thread before any CA calls.
+	e.attachCAContext()
+
 	handle := nextHandle()
 
 	// Register the handle → channel mapping before creating the CA channel so
@@ -129,7 +152,8 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour
 	connTable[handle] = entry
 	handleMu.Unlock()
 
-	// Create the CA channel.
+	// Create the CA channel — this call returns immediately; the connection
+	// callback fires asynchronously on CA's internal thread.
 	pvName := C.CString(signal)
 	defer C.free(unsafe.Pointer(pvName))
 
@@ -142,54 +166,78 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour
 		handleMu.Unlock()
 		return nil, fmt.Errorf("epics: ca_create_channel(%q) failed: status %d", signal, int(status))
 	}
-
-	// Flush the create request and wait for the connection callback.
 	C.ca_flush_io()
 
-	select {
-	case <-entry.connCh:
-		// Connected.
-	case <-time.After(5 * time.Second):
-		// Timeout; clean up and report error.
-		C.ca_clear_channel(chid)
-		handleMu.Lock()
-		delete(handleTable, handle)
-		delete(connTable, handle)
-		handleMu.Unlock()
-		return nil, fmt.Errorf("epics: timeout waiting for channel %q to connect", signal)
-	case <-ctx.Done():
-		C.ca_clear_channel(chid)
-		handleMu.Lock()
-		delete(handleTable, handle)
-		delete(connTable, handle)
-		handleMu.Unlock()
-		return nil, ctx.Err()
-	}
-
-	// Determine the native DBR_TIME_* type for the channel.
-	dbrType := nativeTimeType(chid)
-
-	// Add the value monitor.
-	var evid C.evid
-	status = C.caAddMonitor(chid, C.short(dbrType), C.uintptr_t(handle), &evid)
-	if status != C.ECA_NORMAL {
-		C.ca_clear_channel(chid)
-		handleMu.Lock()
-		delete(handleTable, handle)
-		delete(connTable, handle)
-		handleMu.Unlock()
-		return nil, fmt.Errorf("epics: ca_add_event(%q) failed: status %d", signal, int(status))
-	}
-	C.ca_flush_io()
-
-	caCh := &caChannel{chid: chid, evid: evid, handle: handle}
-
+	// Track the channel immediately so Write can find the chid.
 	e.mu.Lock()
-	e.channels[signal] = caCh
+	e.channels[signal] = &caChannel{chid: chid, handle: handle}
 	e.mu.Unlock()
 
-	cancel := datasource.CancelFunc(func() {
-		if evid != nil {
+	// Use a done channel to signal the goroutine to stop.
+	done := make(chan struct{})
+	var closeOnce sync.Once
+	stopFn := func() { closeOnce.Do(func() { close(done) }) }
+
+	// Background goroutine: wait for connection, then install the monitor.
+	go func() {
+		// Attach CA context to this OS thread before making any CA calls.
+		e.attachCAContext()
+
+		var evid C.evid
+		connected := false
+
+		select {
+		case <-entry.connCh:
+			connected = true
+		case <-time.After(60 * time.Second):
+			// IOC unreachable; report bad quality so the widget shows an error state.
+			select {
+			case ch <- datasource.Value{Timestamp: time.Now(), Data: float64(0), Quality: datasource.QualityBad}:
+			default:
+			}
+		case <-ctx.Done():
+		case <-done:
+		}
+
+		if connected {
+			// Refresh metadata now that the channel is connected so type, unit,
+			// and display limits are accurate (the earlier sendMeta call may have
+			// run before the channel was up, leaving an incomplete cache entry).
+			meta := e.fetchMetadata(chid, signal)
+			e.mu.Lock()
+			e.metadata[signal] = meta
+			e.mu.Unlock()
+
+			// Signal to the broker/dispatcher that metadata has been refreshed.
+			select {
+			case ch <- datasource.Value{MetaUpdate: true}:
+			default:
+			}
+
+			dbrType := nativeTimeType(chid)
+			st := C.caAddMonitor(chid, C.short(dbrType), C.uintptr_t(handle), &evid)
+			if st != C.ECA_NORMAL {
+				select {
+				case ch <- datasource.Value{Timestamp: time.Now(), Data: float64(0), Quality: datasource.QualityBad}:
+				default:
+				}
+				connected = false
+			} else {
+				C.ca_flush_io()
+				e.mu.Lock()
+				e.channels[signal] = &caChannel{chid: chid, evid: evid, handle: handle}
+				e.mu.Unlock()
+			}
+		}
+
+		// Wait until the subscription is cancelled.
+		select {
+		case <-done:
+		case <-ctx.Done():
+		}
+
+		// Cleanup.
+		if connected && evid != nil {
 			C.ca_clear_event(evid)
 		}
 		C.ca_clear_channel(chid)
@@ -203,15 +251,18 @@ func (e *EPICS) Subscribe(ctx context.Context, signal string, ch chan<- datasour
 		e.mu.Lock()
 		delete(e.channels, signal)
 		e.mu.Unlock()
-	})
+	}()
 
-	return cancel, nil
+	return datasource.CancelFunc(stopFn), nil
 }
 
 // GetMetadata performs a synchronous ca_get for DBR_CTRL_DOUBLE (or the
 // appropriate control type) to retrieve units, display limits, enum strings,
 // and writability information for the named signal.
 func (e *EPICS) GetMetadata(ctx context.Context, signal string) (datasource.Metadata, error) {
+	// Attach the CA context to this OS thread before any CA calls.
+	e.attachCAContext()
+
 	// Check cache first.
 	e.mu.Lock()
 	if m, ok := e.metadata[signal]; ok {
@@ -289,7 +340,10 @@ func (e *EPICS) GetMetadata(ctx context.Context, signal string) (datasource.Meta
 
 // fetchMetadata retrieves metadata from a connected chid using ca_get.
 func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata {
-	meta := datasource.Metadata{Name: name}
+	// In EPICS, write access is governed by CA security (host/user rules), not
+	// by the field type.  Default to writable=true; the IOC will reject puts
+	// that violate its security policy.
+	meta := datasource.Metadata{Name: name, Writable: true}
 
 	fieldType := C.ca_field_type(chid)
 	count := C.ca_element_count(chid)
@@ -326,7 +380,7 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata {
 	switch ctrlType {
 	case C.DBR_CTRL_DOUBLE:
 		var buf C.struct_dbr_ctrl_double
-		status := C.ca_get(C.DBR_CTRL_DOUBLE, chid, unsafe.Pointer(&buf))
+		status := C.caGet(C.DBR_CTRL_DOUBLE, chid, unsafe.Pointer(&buf))
 		if status == C.ECA_NORMAL {
 			C.ca_pend_io(3.0)
 			meta.Unit = C.GoString((*C.char)(unsafe.Pointer(&buf.units[0])))
@@ -334,14 +388,11 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata {
 			meta.DisplayHigh = float64(buf.upper_disp_limit)
 			meta.DriveLow = float64(buf.lower_ctrl_limit)
 			meta.DriveHigh = float64(buf.upper_ctrl_limit)
-			// CA does not expose a writable flag directly; assume writable unless
-			// it is a read-only field type.  Conservatively mark as writable.
-			meta.Writable = true
 		}
 
 	case C.DBR_CTRL_LONG:
 		var buf C.struct_dbr_ctrl_long
-		status := C.ca_get(C.DBR_CTRL_LONG, chid, unsafe.Pointer(&buf))
+		status := C.caGet(C.DBR_CTRL_LONG, chid, unsafe.Pointer(&buf))
 		if status == C.ECA_NORMAL {
 			C.ca_pend_io(3.0)
 			meta.Unit = C.GoString((*C.char)(unsafe.Pointer(&buf.units[0])))
@@ -349,12 +400,11 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata {
 			meta.DisplayHigh = float64(buf.upper_disp_limit)
 			meta.DriveLow = float64(buf.lower_ctrl_limit)
 			meta.DriveHigh = float64(buf.upper_ctrl_limit)
-			meta.Writable = true
 		}
 
 	case C.DBR_CTRL_ENUM:
 		var buf C.struct_dbr_ctrl_enum
-		status := C.ca_get(C.DBR_CTRL_ENUM, chid, unsafe.Pointer(&buf))
+		status := C.caGet(C.DBR_CTRL_ENUM, chid, unsafe.Pointer(&buf))
 		if status == C.ECA_NORMAL {
 			C.ca_pend_io(3.0)
 			n := int(buf.no_str)
@@ -363,28 +413,39 @@ func (e *EPICS) fetchMetadata(chid C.chid, name string) datasource.Metadata {
 				strs[i] = C.GoString((*C.char)(unsafe.Pointer(&buf.strs[i][0])))
 			}
 			meta.EnumStrings = strs
-			meta.Writable = true
 		}
 	}
 
 	return meta
 }
 
-// ListSignals returns cached metadata for all currently connected channels.
-// Full enumeration of all available PVs is deferred to Phase 9.
+// ListSignals returns metadata for all currently tracked channels.
+// Channels with cached metadata return full info; others return a stub entry
+// with the PV name so the signal tree can display them immediately after
+// a manual add, before GetMetadata has been called.
 func (e *EPICS) ListSignals(_ context.Context) ([]datasource.Metadata, error) {
 	e.mu.Lock()
 	defer e.mu.Unlock()
 
-	out := make([]datasource.Metadata, 0, len(e.metadata))
-	for _, m := range e.metadata {
+	// Start with all fully-fetched metadata entries.
+	out := make([]datasource.Metadata, 0, len(e.channels))
+	seen := make(map[string]bool, len(e.metadata))
+	for name, m := range e.metadata {
 		out = append(out, m)
+		seen[name] = true
+	}
+	// Add stub entries for channels that are tracked but not yet fetched.
+	for name := range e.channels {
+		if !seen[name] {
+			out = append(out, datasource.Metadata{Name: name, Type: datasource.TypeFloat64})
+		}
 	}
 	return out, nil
 }
 
 // Write puts a new value onto a CA channel.
 func (e *EPICS) Write(_ context.Context, signal string, value any) error {
+	e.attachCAContext()
 	e.mu.Lock()
 	caCh, ok := e.channels[signal]
 	e.mu.Unlock()
@@ -396,20 +457,20 @@ func (e *EPICS) Write(_ context.Context, signal string, value any) error {
 	switch v := value.(type) {
 	case float64:
 		cv := C.double(v)
-		status = C.ca_put(C.DBR_DOUBLE, caCh.chid, unsafe.Pointer(&cv))
+		status = C.caPut(C.DBR_DOUBLE, caCh.chid, unsafe.Pointer(&cv))
 	case int64:
 		cv := C.long(v)
-		status = C.ca_put(C.DBR_LONG, caCh.chid, unsafe.Pointer(&cv))
+		status = C.caPut(C.DBR_LONG, caCh.chid, unsafe.Pointer(&cv))
 	case string:
 		cs := C.CString(v)
 		defer C.free(unsafe.Pointer(cs))
-		status = C.ca_put(C.DBR_STRING, caCh.chid, unsafe.Pointer(cs))
+		status = C.caPut(C.DBR_STRING, caCh.chid, unsafe.Pointer(cs))
 	case bool:
 		var iv C.short
 		if v {
 			iv = 1
 		}
-		status = C.ca_put(C.DBR_SHORT, caCh.chid, unsafe.Pointer(&iv))
+		status = C.caPut(C.DBR_SHORT, caCh.chid, unsafe.Pointer(&iv))
 	default:
 		return fmt.Errorf("epics: unsupported value type %T for Write", value)
 	}

internal/datasource/iface.go

@@ -42,9 +42,10 @@ const (
 
 // Value is a timestamped signal reading.
 type Value struct {
-	Timestamp time.Time
-	Data      any // float64 | []float64 | string | int64 | bool
-	Quality   Quality
+	Timestamp  time.Time
+	Data       any // float64 | []float64 | string | int64 | bool
+	Quality    Quality
+	MetaUpdate bool // if true, metadata was refreshed — dispatcher should re-send meta
 }
 
 // Metadata describes the static properties of a signal.