Documentation
¶
Index ¶
- Constants
- Variables
- type PipelineSpool
- func (ps *PipelineSpool) Abort()
- func (ps *PipelineSpool) Close()
- func (ps *PipelineSpool) CloseWithTimeout(timeout time.Duration) bool
- func (ps *PipelineSpool) ForceCleanup()
- func (ps *PipelineSpool) ForceCleanupAfterTerminalSignal()
- func (ps *PipelineSpool) ReceiveBatch(idx int) (data *batch.Batch, info error)
- func (ps *PipelineSpool) ReleaseCurrent(idx int)
- func (ps *PipelineSpool) SendBatch(ctx context.Context, receiverID int, data *batch.Batch, info error) (bool, error)
Constants ¶
const ( // SendToAllLocal and SendToAnyLocal // are Special receiver IDs for SendBatch method. SendToAllLocal = -1 SendToAnyLocal = -2 )
Variables ¶
var ErrPipelineSpoolAborted = moerr.NewInternalErrorNoCtx("pipeline spool aborted")
Functions ¶
This section is empty.
Types ¶
type PipelineSpool ¶
type PipelineSpool struct {
// contains filtered or unexported fields
}
func InitMyPipelineSpool ¶
func InitMyPipelineSpool(mp *mpool.MPool, receiverCnt uint32) *PipelineSpool
InitMyPipelineSpool return a simple pipeline spool for temporary plan.
todo: use spool package after pipeline construct process is simple.
func (*PipelineSpool) Abort ¶
func (ps *PipelineSpool) Abort()
Abort terminates the spool without waiting for receiver acknowledgement. Pending, not-yet-consumed slots are released immediately. Slots already handed to receivers stay valid until their receiver calls ReleaseCurrent.
func (*PipelineSpool) Close ¶
func (ps *PipelineSpool) Close()
Close the sender and receivers, and do memory clean.
func (*PipelineSpool) CloseWithTimeout ¶
func (ps *PipelineSpool) CloseWithTimeout(timeout time.Duration) bool
func (*PipelineSpool) ForceCleanup ¶
func (ps *PipelineSpool) ForceCleanup()
ForceCleanup reclaims spool-owned batch memory during query teardown, but only once every receiver is confirmed done.
Delivery through the spool is two-staged: SendBatch first stores a batch in a slot, and only then is a GetFromSpool signal enqueued on the receiver channel. A receiver that is merely backlogged during teardown can still hold a pending GetFromSpool signal pointing at a slot with real data. Freeing the backing buffer then (cache.free releases the whole buffer) would make that receiver later read emptied memory -> silent batch loss / early EOS.
So we reclaim only after draining one csDoneSignal per receiver (each receiver emits it once it has read its End-message and therefore released every real-data slot it consumed). If some receiver has not finished yet, we leave the spool memory intact - it is bounded and reclaimed when the owning MPool is destroyed - and do NOT consume cleanupOnce, so a later call can still reclaim once the spool has fully drained.
func (*PipelineSpool) ForceCleanupAfterTerminalSignal ¶
func (ps *PipelineSpool) ForceCleanupAfterTerminalSignal()
ForceCleanupAfterTerminalSignal reclaims cache memory after a typed terminal signal has been delivered outside the spool.
Typed terminal receivers do not consume a nil End-message from the spool, so they never write csDoneSignal. The caller must invoke this only after the paired receiver cleanup loop has returned. On the normal path that loop drains queued GetFromSpool signals; on the timeout path it releases its current batch reference and exits, so no receiver goroutine can later read the pending signals.
func (*PipelineSpool) ReceiveBatch ¶
func (ps *PipelineSpool) ReceiveBatch(idx int) (data *batch.Batch, info error)
ReceiveBatch get data from the idx-th receiver.
func (*PipelineSpool) ReleaseCurrent ¶
func (ps *PipelineSpool) ReleaseCurrent(idx int)
ReleaseCurrent force to release the last received one.
Source Files