README
¶
ndn-dpdk/app/fetch
This package is the congestion aware fetcher, used in the traffic generator.
It implements a consumer that follows the TCP CUBIC congestion control algorithm, simulating traffic patterns similar to bulk file transfer.
It requires at least one thread, running the FetchThread_Run function.
Documentation
¶
Overview ¶
Package fetch retrieves segmented objects.
Index ¶
- Variables
- type Config
- type Counters
- type Fetcher
- func (fetcher *Fetcher) Close() error
- func (fetcher *Fetcher) ConnectRxQueues(demuxD, demuxN *iface.InputDemux)
- func (fetcher *Fetcher) Face() iface.Face
- func (fetcher *Fetcher) Fetch(d TaskDef) (task *TaskContext, e error)
- func (fetcher *Fetcher) Launch()
- func (fetcher *Fetcher) Reset()
- func (fetcher *Fetcher) Stop() error
- func (fetcher *Fetcher) Tasks() (list []*TaskContext)
- func (fetcher *Fetcher) Workers() []ealthread.ThreadWithRole
- type Logic
- type TaskContext
- type TaskDef
- type TaskSlotConfig
Constants ¶
This section is empty.
Variables ¶
var ( GqlConfigInput *graphql.InputObject GqlTaskDefInput *graphql.InputObject GqlTaskDefType *graphql.Object GqlTaskContextType *gqlserver.NodeType[*TaskContext] GqlFetcherType *gqlserver.NodeType[*Fetcher] )
GraphQL types.
var GqlRetrieveByFaceID func(id iface.ID) *Fetcher
GqlRetrieveByFaceID returns *Fetcher associated with a face. It is assigned during package tg initialization.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
TaskSlotConfig
// NThreads is the number of worker threads.
// Each worker thread can serve multiple fetch tasks.
NThreads int `json:"nThreads,omitempty"`
// NTasks is the number of task slots.
// Each task retrieves one segmented object and has independent congestion control.
NTasks int `json:"nTasks,omitempty"`
}
Config contains Fetcher configuration.
type Counters ¶
type Counters struct {
Elapsed time.Duration `json:"elapsed" gqldesc:"Duration since start fetching."`
Finished *time.Duration `json:"finished" gqldesc:"Duration between start and finish; null if not finished."`
LastRtt time.Duration `json:"lastRtt" gqldesc:"Last RTT sample."`
SRtt time.Duration `json:"sRtt" gqldesc:"Smoothed RTT."`
Rto time.Duration `json:"rto" gqldesc:"RTO."`
Cwnd int `json:"cwnd" gqldesc:"Congestion window."`
NInFlight uint32 `json:"nInFlight" gqldesc:"Currently in-flight Interests."`
NTxRetx uint64 `json:"nTxRetx" gqldesc:"Retransmitted Interests."`
NRxData uint64 `json:"nRxData" gqldesc:"Data satisfying pending Interests."`
}
Counters contains counters of Logic.
type Fetcher ¶
type Fetcher struct {
// contains filtered or unexported fields
}
Fetcher controls worker threads and task slots on a face.
func (*Fetcher) ConnectRxQueues ¶
func (fetcher *Fetcher) ConnectRxQueues(demuxD, demuxN *iface.InputDemux)
ConnectRxQueues connects Data InputDemux to RxQueues. Nack InputDemux is set to drop packets because fetcher does not support Nacks.
func (*Fetcher) Fetch ¶
func (fetcher *Fetcher) Fetch(d TaskDef) (task *TaskContext, e error)
Fetch starts a fetch task.
func (*Fetcher) Reset ¶
func (fetcher *Fetcher) Reset()
Reset aborts all tasks and stops all worker threads.
func (*Fetcher) Tasks ¶
func (fetcher *Fetcher) Tasks() (list []*TaskContext)
Tasks returns running fetch tasks.
func (*Fetcher) Workers ¶
func (fetcher *Fetcher) Workers() []ealthread.ThreadWithRole
Workers returns worker threads.
type Logic ¶
type Logic C.FetchLogic
Logic implements fetcher congestion control and scheduling logic.
func (*Logic) Init ¶
func (fl *Logic) Init(winCapacity int, socket eal.NumaSocket)
Init initializes the logic and allocates data structures.
func (*Logic) Reset ¶
func (fl *Logic) Reset(r segmented.SegmentRange)
Reset resets this to initial state.
type TaskContext ¶
type TaskContext struct {
// contains filtered or unexported fields
}
TaskContext provides contextual information about an active fetch task.
func (*TaskContext) Counters ¶
func (task *TaskContext) Counters() Counters
Counters returns congestion control and scheduling counters.
func (*TaskContext) Finished ¶
func (task *TaskContext) Finished() bool
Finished determines if all segments have been fetched.
func (*TaskContext) Stop ¶
func (task *TaskContext) Stop()
Stop aborts/stops the fetch task. This should be called even if the fetch task has succeeded.
type TaskDef ¶
type TaskDef struct {
// InterestTemplateConfig contains the name prefix, InterestLifetime, etc.
//
// The fetcher neither retrieves metadata nor performs version discovery.
// If the content is published with version component, it should appear in the name prefix.
//
// CanBePrefix and MustBeFresh are not normally used, but they may be included for benchmarking purpose.
ndni.InterestTemplateConfig
// SegmentRange specifies range of segment numbers.
// If writing to a file, SegmentEnd must be explicitly specified.
segmented.SegmentRange
// Filename is the output file name.
// If omitted, payload is not written to a file.
Filename string `json:"filename,omitempty"`
// SegmentLen is the payload length in each segment.
// This is only needed when writing to a file.
// If any segment has incorrect Content TLV-LENGTH, the output file would not contain correct payload.
SegmentLen int `json:"segmentLen,omitempty"`
}
TaskDef defines a fetch task that retrieves one segmented object.
type TaskSlotConfig ¶
type TaskSlotConfig struct {
// RxQueue configures the RX queue of Data packets going to each task slot.
// CoDel cannot be used in these queues.
RxQueue iface.PktQueueConfig `json:"rxQueue,omitempty"`
// WindowCapacity is the maximum distance between lower and upper bounds of segment numbers in an ongoing fetch logic.
WindowCapacity int `json:"windowCapacity,omitempty"`
}
TaskSlotConfig contains task slot configuration.
Source Files
Directories