mirror of
https://github.com/router-for-me/CLIProxyAPI.git
synced 2026-06-23 20:04:08 +08:00
Treat loading plugins as busy
This commit is contained in:
@@ -198,15 +198,15 @@ func (h *Handler) installPluginFromStore(c *gin.Context, goos, goarch string) {
|
||||
return
|
||||
}
|
||||
|
||||
pluginIsLoaded := func() bool { return pluginLoaded(host, id) }
|
||||
pluginIsBusy := func() bool { return pluginBusy(host, id) }
|
||||
unloadedBeforeWrite := false
|
||||
result, errInstall := client.Install(installCtx, plugin, pluginstore.InstallOptions{
|
||||
PluginsDir: pluginsDir,
|
||||
GOOS: goos,
|
||||
GOARCH: goarch,
|
||||
PluginLoaded: pluginIsLoaded,
|
||||
PluginLoaded: pluginIsBusy,
|
||||
BeforeWrite: func() error {
|
||||
if !pluginIsLoaded() {
|
||||
if !pluginIsBusy() {
|
||||
return nil
|
||||
}
|
||||
if host == nil {
|
||||
@@ -215,8 +215,8 @@ func (h *Handler) installPluginFromStore(c *gin.Context, goos, goarch string) {
|
||||
log.WithFields(log.Fields{
|
||||
"plugin_id": id,
|
||||
"version": plugin.Version,
|
||||
}).Info("pluginstore: unloading loaded plugin before install")
|
||||
if !host.UnloadPlugin(id) && pluginIsLoaded() {
|
||||
}).Info("pluginstore: unloading busy plugin before install")
|
||||
if !host.UnloadPlugin(id) && pluginIsBusy() {
|
||||
return pluginstore.ErrLoadedPluginLocked
|
||||
}
|
||||
unloadedBeforeWrite = true
|
||||
@@ -560,9 +560,9 @@ func pluginLocalStatuses(pluginsEnabled bool, pluginsDir string, configs map[str
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func pluginLoaded(host *pluginhost.Host, id string) bool {
|
||||
func pluginBusy(host *pluginhost.Host, id string) bool {
|
||||
if host == nil {
|
||||
return false
|
||||
}
|
||||
return host.PluginLoaded(id)
|
||||
return host.PluginBusy(id)
|
||||
}
|
||||
|
||||
@@ -338,7 +338,7 @@ func (h *Handler) DeletePlugin(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
if pluginLoaded(host, id) && (host == nil || !host.UnloadPlugin(id)) && pluginLoaded(host, id) {
|
||||
if pluginBusy(host, id) && (host == nil || !host.UnloadPlugin(id)) && pluginBusy(host, id) {
|
||||
c.JSON(http.StatusConflict, gin.H{
|
||||
"error": "plugin_delete_requires_restart",
|
||||
"message": "loaded plugin cannot be deleted while the server is running",
|
||||
|
||||
@@ -39,6 +39,7 @@ type Host struct {
|
||||
mu sync.Mutex
|
||||
loader pluginLoader
|
||||
loaded map[string]*loadedPlugin
|
||||
loading map[string]struct{}
|
||||
fused map[string]string
|
||||
runtimeConfig *config.Config
|
||||
authManager *coreauth.Manager
|
||||
@@ -65,6 +66,7 @@ func New() *Host {
|
||||
h := &Host{
|
||||
loader: defaultPluginLoader(),
|
||||
loaded: make(map[string]*loadedPlugin),
|
||||
loading: make(map[string]struct{}),
|
||||
fused: make(map[string]string),
|
||||
modelClientIDs: make(map[string]struct{}),
|
||||
executorModelClientIDs: make(map[string]struct{}),
|
||||
@@ -137,6 +139,24 @@ func (h *Host) PluginLoaded(id string) bool {
|
||||
return ok
|
||||
}
|
||||
|
||||
// PluginBusy reports whether a plugin dynamic library is loaded or being loaded.
|
||||
func (h *Host) PluginBusy(id string) bool {
|
||||
if h == nil {
|
||||
return false
|
||||
}
|
||||
id = strings.TrimSpace(id)
|
||||
if id == "" {
|
||||
return false
|
||||
}
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if _, ok := h.loaded[id]; ok {
|
||||
return true
|
||||
}
|
||||
_, ok := h.loading[id]
|
||||
return ok
|
||||
}
|
||||
|
||||
func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
|
||||
if h == nil {
|
||||
return
|
||||
@@ -189,12 +209,18 @@ func (h *Host) ApplyConfig(ctx context.Context, cfg *config.Config) {
|
||||
}
|
||||
|
||||
if lp == nil {
|
||||
h.mu.Lock()
|
||||
h.loading[file.ID] = struct{}{}
|
||||
h.mu.Unlock()
|
||||
|
||||
loaded, errLoad := h.load(file)
|
||||
h.mu.Lock()
|
||||
delete(h.loading, file.ID)
|
||||
if errLoad != nil {
|
||||
h.mu.Unlock()
|
||||
log.Warnf("pluginhost: failed to load plugin %s from %s: %v", file.ID, file.Path, errLoad)
|
||||
continue
|
||||
}
|
||||
h.mu.Lock()
|
||||
// ApplyConfig, UnloadPlugin, and ShutdownAll are serialized by applyMu,
|
||||
// so a nil read cannot race into a duplicate load.
|
||||
lp = loaded
|
||||
@@ -301,6 +327,7 @@ func (h *Host) ShutdownAll() {
|
||||
})
|
||||
}
|
||||
h.loaded = make(map[string]*loadedPlugin)
|
||||
h.loading = make(map[string]struct{})
|
||||
h.modelClientIDs = make(map[string]struct{})
|
||||
h.executorModelClientIDs = make(map[string]struct{})
|
||||
h.modelProviders = make(map[string]string)
|
||||
|
||||
@@ -707,6 +707,63 @@ func TestHostApplyConfigSerializesLifecycleCalls(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostPluginBusyReportsLoadingPlugin(t *testing.T) {
|
||||
h, cfg, openStarted, releaseOpen := newBlockingOpenHost(t)
|
||||
t.Cleanup(h.ShutdownAll)
|
||||
|
||||
applyDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(applyDone)
|
||||
}()
|
||||
|
||||
waitForHostTestSignal(t, openStarted, "plugin open start")
|
||||
if h.PluginLoaded("alpha") {
|
||||
t.Fatal("PluginLoaded(alpha) = true, want false while plugin is still loading")
|
||||
}
|
||||
if !h.PluginBusy("alpha") {
|
||||
t.Fatal("PluginBusy(alpha) = false, want true while plugin is loading")
|
||||
}
|
||||
|
||||
releaseOpen()
|
||||
waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
|
||||
if !h.PluginLoaded("alpha") {
|
||||
t.Fatal("PluginLoaded(alpha) = false, want true after load")
|
||||
}
|
||||
if !h.PluginBusy("alpha") {
|
||||
t.Fatal("PluginBusy(alpha) = false, want true after load")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostUnloadWaitsForBlockingLoad(t *testing.T) {
|
||||
h, cfg, openStarted, releaseOpen := newBlockingOpenHost(t)
|
||||
applyDone := make(chan struct{})
|
||||
go func() {
|
||||
h.ApplyConfig(context.Background(), cfg)
|
||||
close(applyDone)
|
||||
}()
|
||||
waitForHostTestSignal(t, openStarted, "plugin open start")
|
||||
|
||||
unloadDone := make(chan bool)
|
||||
go func() {
|
||||
unloadDone <- h.UnloadPlugin("alpha")
|
||||
}()
|
||||
select {
|
||||
case <-unloadDone:
|
||||
t.Fatal("UnloadPlugin completed while ApplyConfig was still loading")
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
|
||||
releaseOpen()
|
||||
waitForHostTestSignal(t, applyDone, "ApplyConfig completion")
|
||||
if ok := waitForHostTestBool(t, unloadDone, "UnloadPlugin completion"); !ok {
|
||||
t.Fatal("UnloadPlugin returned false, want true after loading completes")
|
||||
}
|
||||
if h.PluginBusy("alpha") {
|
||||
t.Fatal("PluginBusy(alpha) = true, want false after unload")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHostUnloadAndShutdownWaitForBlockingRegister(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
@@ -801,6 +858,49 @@ func (c *capturePluginClient) Call(ctx context.Context, method string, request [
|
||||
|
||||
func (c *capturePluginClient) Shutdown() {}
|
||||
|
||||
type blockingOpenLoader struct {
|
||||
inner *testSymbolLoader
|
||||
started chan struct{}
|
||||
release <-chan struct{}
|
||||
startOnce sync.Once
|
||||
}
|
||||
|
||||
func (l *blockingOpenLoader) Open(file pluginFile, host *Host) (pluginClient, error) {
|
||||
l.startOnce.Do(func() { close(l.started) })
|
||||
<-l.release
|
||||
return l.inner.Open(file, host)
|
||||
}
|
||||
|
||||
func newBlockingOpenHost(t *testing.T) (*Host, *config.Config, <-chan struct{}, func()) {
|
||||
t.Helper()
|
||||
|
||||
inner := newTestSymbolLoader()
|
||||
plugin := &testPlugin{
|
||||
registerResult: validTestPlugin("alpha"),
|
||||
reconfigureResult: validTestPlugin("alpha"),
|
||||
}
|
||||
inner.lookups["alpha"] = newTestSymbolLookup(plugin)
|
||||
|
||||
openStarted := make(chan struct{})
|
||||
release := make(chan struct{})
|
||||
var releaseOnce sync.Once
|
||||
releaseOpen := func() { releaseOnce.Do(func() { close(release) }) }
|
||||
t.Cleanup(releaseOpen)
|
||||
|
||||
h := NewForTest(&blockingOpenLoader{
|
||||
inner: inner,
|
||||
started: openStarted,
|
||||
release: release,
|
||||
})
|
||||
cfg := &config.Config{
|
||||
Plugins: config.PluginsConfig{
|
||||
Enabled: true,
|
||||
Dir: makePluginDir(t, "alpha"),
|
||||
},
|
||||
}
|
||||
return h, cfg, openStarted, releaseOpen
|
||||
}
|
||||
|
||||
func newBlockingRegisterHost(t *testing.T) (*Host, *config.Config, <-chan struct{}, func()) {
|
||||
t.Helper()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user