|
| 1 | +package galleryop |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "fmt" |
| 6 | + |
| 7 | + "github.com/mudler/LocalAI/core/gallery" |
| 8 | + "github.com/mudler/xlog" |
| 9 | +) |
| 10 | + |
| 11 | +// queuedMessage is the status message an operation carries between admission |
| 12 | +// (the HTTP handler minting the job ID) and the moment the worker starts it. |
| 13 | +const queuedMessage = "queued" |
| 14 | + |
| 15 | +// EnqueueModelOp admits a model operation: it registers a queryable "queued" |
| 16 | +// status for op.ID and then hands the op to the gallery worker. |
| 17 | +// |
| 18 | +// Why the status is written here and not in modelHandler: the gallery channels |
| 19 | +// are unbuffered and the worker processes one operation at a time, so an op |
| 20 | +// submitted while another install is downloading sits in a blocked send for as |
| 21 | +// long as that install takes. The admission handlers return HTTP 200 with the |
| 22 | +// job ID immediately, so for that whole window the client held an ID that |
| 23 | +// GET /models/jobs/<id> answered with "could not find any status for ID" and |
| 24 | +// GET /models/jobs did not list at all — the endpoint reported success for work |
| 25 | +// nothing could observe. Writing the status before the send makes the job |
| 26 | +// queryable from the instant its ID is handed out. |
| 27 | +// |
| 28 | +// Delivery is asynchronous on purpose: a direct send would block the HTTP |
| 29 | +// handler for the entire duration of the in-flight install (observed as a |
| 30 | +// request that never gets a response while /readyz stays green). |
| 31 | +func (g *GalleryService) EnqueueModelOp(op ManagementOp[gallery.GalleryModel, gallery.ModelConfig]) { |
| 32 | + g.markQueued(op.ID, op.GalleryElementName, op.Delete) |
| 33 | + go func() { |
| 34 | + select { |
| 35 | + case g.ModelGalleryChannel <- op: |
| 36 | + case <-enqueueContext(op.Context).Done(): |
| 37 | + g.abandonQueued(op.ID, op.GalleryElementName) |
| 38 | + } |
| 39 | + }() |
| 40 | +} |
| 41 | + |
| 42 | +// EnqueueBackendOp is the BackendGalleryChannel sibling of EnqueueModelOp. |
| 43 | +// Same rationale — see that comment. |
| 44 | +func (g *GalleryService) EnqueueBackendOp(op ManagementOp[gallery.GalleryBackend, any]) { |
| 45 | + g.markQueued(op.ID, op.GalleryElementName, op.Delete) |
| 46 | + go func() { |
| 47 | + select { |
| 48 | + case g.BackendGalleryChannel <- op: |
| 49 | + case <-enqueueContext(op.Context).Done(): |
| 50 | + g.abandonQueued(op.ID, op.GalleryElementName) |
| 51 | + } |
| 52 | + }() |
| 53 | +} |
| 54 | + |
| 55 | +// enqueueContext gives the delivery goroutine something to select on. Ops |
| 56 | +// submitted without a context (the /models/apply path) simply wait for the |
| 57 | +// worker; ops that carry one (every UI path) are released the moment the |
| 58 | +// operation is cancelled, so cancelling a still-queued op no longer strands |
| 59 | +// the goroutine on a send that will never be received. |
| 60 | +func enqueueContext(ctx context.Context) context.Context { |
| 61 | + if ctx == nil { |
| 62 | + return context.Background() |
| 63 | + } |
| 64 | + return ctx |
| 65 | +} |
| 66 | + |
| 67 | +// markQueued publishes the pre-worker status for an admitted operation. |
| 68 | +func (g *GalleryService) markQueued(id, elementName string, deletion bool) { |
| 69 | + if id == "" { |
| 70 | + return |
| 71 | + } |
| 72 | + g.UpdateStatus(id, &OpStatus{ |
| 73 | + Message: queuedMessage, |
| 74 | + Phase: queuedMessage, |
| 75 | + GalleryElementName: elementName, |
| 76 | + Deletion: deletion, |
| 77 | + Cancellable: !deletion, |
| 78 | + }) |
| 79 | +} |
| 80 | + |
| 81 | +// abandonQueued turns a queued operation that the worker never accepted into a |
| 82 | +// terminal failure. Without it the op keeps claiming it is waiting to start |
| 83 | +// while the goroutine that was supposed to deliver it is gone, which is the |
| 84 | +// same phantom-operation shape as an op orphaned by a replica restart. |
| 85 | +// |
| 86 | +// A status that already reached a terminal state (a concurrent cancel is the |
| 87 | +// common case) wins: we must not overwrite "cancelled" with a generic failure. |
| 88 | +func (g *GalleryService) abandonQueued(id, elementName string) { |
| 89 | + if id == "" { |
| 90 | + return |
| 91 | + } |
| 92 | + g.Lock() |
| 93 | + if st, ok := g.statuses[id]; ok && st.Processed { |
| 94 | + g.Unlock() |
| 95 | + return |
| 96 | + } |
| 97 | + g.Unlock() |
| 98 | + |
| 99 | + xlog.Warn("Gallery operation was never picked up by the worker", "op_id", id, "element", elementName) |
| 100 | + g.UpdateStatus(id, &OpStatus{ |
| 101 | + Processed: true, |
| 102 | + Error: fmt.Errorf("operation was cancelled before the gallery worker could start it"), |
| 103 | + Message: "error: operation was cancelled before the gallery worker could start it", |
| 104 | + GalleryElementName: elementName, |
| 105 | + }) |
| 106 | +} |
0 commit comments