kube-controller-manager 分析

发表于 2017-11-27 更新于 2025-03-14 Disqus：

kube-controller-manager 可以认为是一个守护进程用于监视 apiserver 暴露的集群状态，并且不断地尝试把当前状态向集群的目标状态迁移。为了避免频繁查询 apiserver，apiserver 提供了 watch 接口用于监视资源的增加删除和更新，client-go 对此作了抽象，封装一层 informer 来表示本地 apiserver 状态的 cache。这个视频有一个 Google 工程师讲解的 client-go 的详细内容，这篇七牛前同事的文章介绍了 informer 的整体结构，写得也很好。

client 当中的 controller

处理事件的 controller 由几部分构成，首先是 Config 当中的可配置部分，下图是 controller 的关系，controller 实现了 Controller 接口。

controller 从 Queue 当中通过 Pop 获取对象交给 Process 回调处理，DeltaFIFO 和 FIFO 是类似的，只是 DeltaFIFO 可以处理删除事件，一般都用 DeltaFIFO。ListerWatcher 就是用客户端构造出来的，针对对应资源的 List Watch 方法的集合，List 用于获取最开始的对象获取，Watch 用于监控之后的变化，所有最开始的时候现有的对象会通过 List 传给 Add 回调，同步了当前状态以后再不断接受新的变化，但是 Watch 本身是有超时机制的，不能永久监听，所以再超时之后还会通过 List 方法，先同步一次再进行删除操作。Resync Period 表示把 cache 中的对象重新入队给回调函数处理，这种情况一般是可能你可能漏掉了更细操作，或者是之前的一些失败了。大部分情况用不到这个选项，可以非常相信 etcd 的功能。

调用路径

cache.NewListWatchFromClient -> listWatcher
cache.NewIndexer -> store
cache.NewLister(store) -> lister
cache.NewReflector
go refector.Run()

还有一个关键结构是 reflector，reflector 会把对象转化成对应的需要的对象并且 Add 到 Queue 当中去。

Informer 本身的框架是异步的，所以为了做并发控制就引入了 workqueue 的组件，workqueue 有 rate limit 的功能，并且能够合并更新操作。

注意不要修改传入的对象，因为他们要和 cache 一致，如果要写对象的话，需要使用 api.Scheme.Copy 这个函数，进行深度拷贝，所有的 k8s Object 都要支持深拷贝的方法。

kube-controller 当中的插件式 controller

在k8s.io/kubernetes/cmd/kube-controller-manager/app/controllermanager.go 中的NewControllerInitializers 函数有大部分 controller 的列表，bootstrapsigner 和 tokencleaner 是默认关闭的。

var ControllersDisabledByDefault = sets.NewString(
	"bootstrapsigner",
	"tokencleaner",
)

需要特殊初始化的是 serviceaccount-tok，另外的是 NewControllerInitializers 当中的 controller 了。所有的 controller 初始化函数都要满足如下接口。

// InitFunc is used to launch a particular controller.  It may run additional "should I activate checks".
// Any error returned will cause the controller process to `Fatal`
// The bool indicates whether the controller was enabled.
type InitFunc func(ctx ControllerContext) (bool, error)

node controller

接下来看一个具体的 controller，startNodeController。

startNodeController 首先解析 ClusterCIDR 和 ServiceCIDR 两个子网范围，下面是 NodeController 初始化需要的参数。

nodeController, err := nodecontroller.NewNodeController(
	ctx.InformerFactory.Core().V1().Pods(),
	ctx.InformerFactory.Core().V1().Nodes(),
	ctx.InformerFactory.Extensions().V1beta1().DaemonSets(),
	ctx.Cloud,
	ctx.ClientBuilder.ClientOrDie("node-controller"),
	ctx.Options.PodEvictionTimeout.Duration,
	ctx.Options.NodeEvictionRate,
	ctx.Options.SecondaryNodeEvictionRate,
	ctx.Options.LargeClusterSizeThreshold,
	ctx.Options.UnhealthyZoneThreshold,
	ctx.Options.NodeMonitorGracePeriod.Duration,
	ctx.Options.NodeStartupGracePeriod.Duration,
	ctx.Options.NodeMonitorPeriod.Duration,
	clusterCIDR,
	serviceCIDR,
	int(ctx.Options.NodeCIDRMaskSize),
	ctx.Options.AllocateNodeCIDRs,
	ipam.CIDRAllocatorType(ctx.Options.CIDRAllocatorType),
	ctx.Options.EnableTaintManager,
	utilfeature.DefaultFeatureGate.Enabled(features.TaintBasedEvictions),
	utilfeature.DefaultFeatureGate.Enabled(features.TaintNodesByCondition),
)

InformerFactory 是用来构造具体 resource 的 informer 的工厂类型，构造了 pods, nodes, daemonsets 的 informer, 说明 node controller 需要 watch 这几种 resource 的变化。

// Run starts an asynchronous loop that monitors the status of cluster nodes.
func (nc *Controller) Run(stopCh <-chan struct{}) {
        defer utilruntime.HandleCrash()

        glog.Infof("Starting node controller")
        defer glog.Infof("Shutting down node controller")

        if !controller.WaitForCacheSync("node", stopCh, nc.nodeInformerSynced, nc.podInformerSynced, nc.daemonSetInformerSynced) {
                return
        }

        // Incorporate the results of node status pushed from kubelet to master.
        go wait.Until(func() {
                if err := nc.monitorNodeStatus(); err != nil {
                        glog.Errorf("Error monitoring node status: %v", err)
                }
        }, nc.nodeMonitorPeriod, wait.NeverStop)

        if nc.runTaintManager {
                go nc.taintManager.Run(wait.NeverStop)
        }

        if nc.useTaintBasedEvictions {
                // Handling taint based evictions. Because we don't want a dedicated logic in TaintManager for NC-originated
                // taints and we normally don't rate limit evictions caused by taints, we need to rate limit adding taints.
                go wait.Until(nc.doNoExecuteTaintingPass, scheduler.NodeEvictionPeriod, wait.NeverStop)
        } else {
                // Managing eviction of nodes:
                // When we delete pods off a node, if the node was not empty at the time we then
                // queue an eviction watcher. If we hit an error, retry deletion.
                go wait.Until(nc.doEvictionPass, scheduler.NodeEvictionPeriod, wait.NeverStop)
        }

        <-stopCh
}

现在看一下运行的时候是如何做的，首先要调用 controller.WaitForCacheSync 等待 node，pod，daemonSet 的 inofrmer 同步，这是因为在 kube-controller-manager 当中使用的子 controller 使用的 informer 都是共享型的，也就是多个 controller 之间共享一个 informer 的 cache，所以在开始的时候需要保证所有的 sharedInformerFactory 创建的 informers 之间的 cache 先等待一次一致。controller 的读基本上是从 cache 读的，只要写才会打到 etcd 里面，然后等待 cache 的更新回调。

nodeController 主要分成两部分，一部分是 monitorNodeStatus , 它首先从 informer 的 cache 当中 list 新添加的节点和删除的节点，和 newZoneRepresentations。node 是分 zone 的，这在单可用区的 cluster 当中是个空字符串，但是如果 labels 中有 failure-domain.beta.kubernetes.io/zone 和 failure-domain.beta.kubernetes.io/region 就会构成不同的可用区的划分, 这个适用于仅仅在一家云厂商分不同可用区的时候可以用到。这篇文档描述了多 zone 的 cluster 的内容，没有 zone 有相应的可用状态，如果某个 zone 变成不可用需要把 pod 从这个 zone 当中剔除，所以 pod 的 failover 是以 zone 为单位的。

处理 node 比较啰嗦，tryUpdateNodeStatus 尝试获取当前的 conditions 更新并且获取 conditions。处理 node 过程主要是标记 node 为不可用的 node，或者把不可用的状态恢复过来。

接下来的就是处理 pod eviction 的部分，另外，这篇文档解释了一些 kubelet 支持的资源耗尽的情况下 kubelet 的剔除策略。

taintManager.Run 处理 taint 的逻辑，这篇文档解释了 taint 和 toleration 的关系，以及基于 taint 的 eviction 策略。首先看 node 的更新，然后把上面不能 tolerate 的 pod 传给 handlePodUpdate，然后 pod 有更新也会 handlePodUpdate，在 pod 更新的时间中会让 node 抢占一次。

for {
        select {
        case <-stopCh:
                break
        case nodeUpdate := <-tc.nodeUpdateChannel:
                tc.handleNodeUpdate(nodeUpdate)
        case podUpdate := <-tc.podUpdateChannel:
                // If we found a Pod update we need to empty Node queue first.
        priority:
                for {
                        select {
                        case nodeUpdate := <-tc.nodeUpdateChannel:
                                tc.handleNodeUpdate(nodeUpdate)
                        default:
                                break priority
                        }
                }
                // After Node queue is emptied we process podUpdate.
                tc.handlePodUpdate(podUpdate)
        }
}

最后是 eviction 的部分，基于 eviction 的会把 pod 直接删除，基于 taint 只是打上标记，然后通过上面的 tiantManager 剔除。和 pod eviction 不同，taint eviction 是通过限制加 taint 的速率控制 raltelimit 的。

podGCController

pod GC Controller 只 watch pod 一种资源，比较简单。

在 k8s.io/kubernetes/pkg/controller/podgc/gc_controller.go 下。

gcc.gcOrphaned 删除 node 不存在的 pod，gcc.gcUnscheduledTerminating 删除正在终止，但是没有调度的 pod，gcc.gcTerminated 删除已经被终止的 pod。

其他 controller

其他 controller 也是类似的，从 apiserver 获取状态，并且向对应的状态迁移，这也是为什么 kubernetes 的命令和资源都是宣告式的原因。

Kubernetes API 分析

发表于 2017-11-05 更新于 2025-03-14 Disqus：

kubernetes 概览

以下是 k8s 的整体架构，在 master 节点上主要是 kube-apiserver（整合了 kube-aggregator），还有 kube-scheduler，以及 kube-controller-manager，包括后端存储 etcd。

其中 kube-apiserver 是一个比较关键的部分，而且前期写得坑很多，导致这一部分虽然看起来是一个 API server 其实代码很复杂，特别冗余，而且目前对 kube-apiserver 还要做拆分，能够支持插入第三方的 apiserver，也就是又一个 aggregated apiserver 的 feature，也是和 kube-apiserver 和里面包的一层 genericserver 揉合在一起了，感觉一个大的系统 API server 越写越挫是一个通病，还好现在 k8s 迷途知返正在调整。

kube-apiserver

Kube-apiserver 可以是认为在 generic server 上封装的一层官方默认的 apiserver，有第三方需要的情况下，自己也可以在 generic server 上封装一层加入到集成模式中，这里主要介绍 kube-apiserver 的结构。

restful API

kube-apiserver 是一个 restful 服务，请求直接通过 HTTP 请求发送，例如创建一个 ubuntu 的 pod，用以下的 pod.yaml 文件。

apiVersion: v1
kind: Pod
metadata:
  name: ubuntu1
  labels:
    name: ubuntu1
spec:
  containers:
  - name: ubuntu1
    image: ubuntu
    command: ["sleep", "1d"]

执行命令 kubectl create -f ./pod.yaml -v=8，可以看到对应的 POST 请求如下。

Request Body: {"apiVersion":"v1","kind":"Pod","metadata":{"labels":{"name":"ubuntu1"},"name":"ubuntu1","namespace":"default"},"spec":{"containers":[{"command":["sleep","1d"],"image":"ubuntu","name":"ubuntu1"}],"schedulerName":"default-scheduler"}}
curl -k -v -XPOST  -H "Content-Type: application/json" -H "Accept: application/json" -H "User-Agent: kubectl/v1.7.5 (linux/amd64) kubernetes/17d7182" https://localhost:6443/api/v1/namespaces/default/pods
POST https://localhost:6443/api/v1/namespaces/default/pods 201 Created in 6 milliseconds
Response Headers:
    Content-Type: application/json
    Content-Length: 1208
    Date: Wed, 18 Oct 2017 15:04:17 GMT
Response Body: {"kind":"Pod","apiVersion":"v1","metadata":{"name":"ubuntu1","namespace":"default","selfLink":"/api/v1/namespaces/default/pods/ubuntu1","uid":"9c9af581-b415-11e7-8033-024d1ba659e8","resourceVersion":"486154","creationTimestamp":"2017-10-18T15:04:17Z","labels":{"name":"ubuntu1"}},"spec":{"volumes":[{"name":"default-token-p0980","secret":{"secretName":"default-token-p0980","defaultMode":420}}],"containers":[{"name":"ubuntu1","image":"ubuntu","command":["sleep","1d"],"resources":{},"volumeMounts":[{"name":"default-token-p0980","readOnly":true,"mountPath":"/var/run/secrets/kubernetes.io/serviceaccount"}],"terminationMessagePath":"/dev/termination-log","terminationMessagePolicy":"File","imagePullPolicy":"Always"}],"restartPolicy":"Always","terminationGracePeriodSeconds":30,"dnsPolicy":"ClusterFirst","serviceAccountName":"default","serviceAccount":"default","securityContext":{},"schedulerName":"default-scheduler","tolerations":[{"key":"node.kubernetes.io/not-ready","operator":"Exists","effect":"NoExecute","tolerationSeconds":300},{"key":"node.alpha.kubernetes.io/unreachable","operator":"Exists","effect":"NoExecute","tolerationSeconds":300}]},"status":{"phase":"Pending","qosClass":"BestEffort"}}

从 url path 里面可以看到几个划分，path 的分类大概有下面这几种。

路径上整体分成 group, version, resource, 作为核心 API group 的 core（包括 pod, node 之类的 resource)，不带 group，直接接在 /api/ 后面，其他的 api group 则接在 /apis 后面。以 pod 为例，pod 对应的数据类型如下，这个数据结构和 POST 请求中的结构的参数是一致的。

如果是 job 的话则是在，pkg/apis/batch/v2alpha1/types.go，和 API 路径是对应的。例子当中 kubectl 加上 level 大于 8 的 log 就会打印请求和相应的 body，可以看到 request body 和上面的数据结构是一致的。这个请求会发送到 apiserver 进行处理并且返回存储之后的 pod。

重要结构体

Config

父结构，主要的配置内容，其中有一个结构 RESTOptionsGetter genericregistry.RESTOptionsGetter 是和 API 初始化相关的，这个接口的实现是在 k8s.io/apiserver/pkg/server/options/etcd.go 中的 storageFactoryRestOptionsFactory 实现的，对应的实现函数是

func (f *storageFactoryRestOptionsFactory) GetRESTOptions(resource schema.GroupResource) (generic.RESTOptions, error) {
	storageConfig, err := f.StorageFactory.NewConfig(resource)
	if err != nil {
		return generic.RESTOptions{}, fmt.Errorf("unable to find storage destination for %v, due to %v", resource, err.Error())
	}

	ret := generic.RESTOptions{
		StorageConfig:           storageConfig,
		Decorator:               generic.UndecoratedStorage,
		DeleteCollectionWorkers: f.Options.DeleteCollectionWorkers,
		EnableGarbageCollection: f.Options.EnableGarbageCollection,
		ResourcePrefix:          f.StorageFactory.ResourcePrefix(resource),
	}
	if f.Options.EnableWatchCache {
		sizes, err := ParseWatchCacheSizes(f.Options.WatchCacheSizes)
		if err != nil {
			return generic.RESTOptions{}, err
		}
		cacheSize, ok := sizes[resource]
		if !ok {
			cacheSize = f.Options.DefaultWatchCacheSize
		}
		ret.Decorator = genericregistry.StorageWithCacher(cacheSize)
	}

	return ret, nil
}

APIGroupInfo

APIGroupInfo 主要定义了一个 API 组的相关信息，观察一下 APIGroupInfo 是如何初始化的。

在 k8s.io/pkg/master/master.go 当中，每个 Resource 都要提供自己的 Provider，比如说 storagerest 就在 k8s.io/kubernetes/pkg/registry/storage/rest/storage_storage.go 定义了 NewRESTStorage 方法。而默认的 resource 的 legacy provider 单独处理。

if c.ExtraConfig.APIResourceConfigSource.AnyResourcesForVersionEnabled(apiv1.SchemeGroupVersion) {
	legacyRESTStorageProvider := corerest.LegacyRESTStorageProvider{
		StorageFactory:       c.ExtraConfig.StorageFactory,
		ProxyTransport:       c.ExtraConfig.ProxyTransport,
		KubeletClientConfig:  c.ExtraConfig.KubeletClientConfig,
		EventTTL:             c.ExtraConfig.EventTTL,
		ServiceIPRange:       c.ExtraConfig.ServiceIPRange,
		ServiceNodePortRange: c.ExtraConfig.ServiceNodePortRange,
		LoopbackClientConfig: c.GenericConfig.LoopbackClientConfig,
	}
	m.InstallLegacyAPI(&c, c.GenericConfig.RESTOptionsGetter, legacyRESTStorageProvider)
}

然后通过调用 k8s.io/kubernetes/pkg/registry/core/rest.LegacyRESTStorageProvider 的 NewLegacyRESTStorage 来初始化基础对象的 apigroup info，比如初始化 podStorage，serviceStorage 和 nodeStorage 等等。legacy ApiGrouInfo 的 Scheme, ParamaterCodec, NegotiatedSerializer 都是用 "k8s.io/kubernetes/pkg/api" 包下的全局变量初始化的。

1
2
3

Scheme:                      api.Scheme,
ParameterCodec:              api.ParameterCodec,
NegotiatedSerializer:        api.Codecs,

然后合并成一个 restStorage 存入 apiGroupInfo 中。

restStorageMap := map[string]rest.Storage{
	"pods":             podStorage.Pod,
	"pods/attach":      podStorage.Attach,
	"pods/status":      podStorage.Status,
	"pods/log":         podStorage.Log,
	"pods/exec":        podStorage.Exec,
	"pods/portforward": podStorage.PortForward,
	"pods/proxy":       podStorage.Proxy,
	"pods/binding":     podStorage.Binding,
	"bindings":         podStorage.Binding,
   ...

举个例子 podStorage 就是用的 genericregistry.Store，这是一个通用的 etc 辅助结构，把 etcd 抽象成存储结构。

// REST implements a RESTStorage for pods
type REST struct {
	*genericregistry.Store
	proxyTransport http.RoundTripper
}

serialization

pkg/api.Codecs 是全局默认的 codec 来自下面这段代码。

func NewCodecFactory(scheme *runtime.Scheme) CodecFactory {
	serializers := newSerializersForScheme(scheme, json.DefaultMetaFactory)
	return newCodecFactory(scheme, serializers)
}

默认具体定义了这几种 serilizer。

func newSerializersForScheme(scheme *runtime.Scheme, mf json.MetaFactory) []serializerType {
	jsonSerializer := json.NewSerializer(mf, scheme, scheme, false)
	jsonPrettySerializer := json.NewSerializer(mf, scheme, scheme, true)
	yamlSerializer := json.NewYAMLSerializer(mf, scheme, scheme)
    ...

而且标准库的 json 有很严重的性能问题，换用了 json-iter 但是有很多标准库不兼容的问题，性能提升了大概 20% 但是没办法和进主线，我尝试在上面工作的了一段时间，改了两个问题还是有错，由于时间关系，暂时放弃了这个工作，相关的 issue 在这里

filters

首先通过 ./staging/src/k8s.io/apiserver/pkg/server/config.go 下的 DefaultBuildHandlerChain 构建 filters。

func DefaultBuildHandlerChain(apiHandler http.Handler, c *Config) http.Handler {
	handler := genericapifilters.WithAuthorization(apiHandler, c.RequestContextMapper, c.Authorizer, c.Serializer)
	handler = genericfilters.WithMaxInFlightLimit(handler, c.MaxRequestsInFlight, c.MaxMutatingRequestsInFlight, c.RequestContextMapper, c.LongRunningFunc)
	handler = genericapifilters.WithImpersonation(handler, c.RequestContextMapper, c.Authorizer, c.Serializer)
	if utilfeature.DefaultFeatureGate.Enabled(features.AdvancedAuditing) {
		handler = genericapifilters.WithAudit(handler, c.RequestContextMapper, c.AuditBackend, c.AuditPolicyChecker, c.LongRunningFunc)
	} else {
		handler = genericapifilters.WithLegacyAudit(handler, c.RequestContextMapper, c.LegacyAuditWriter)
	}
	failedHandler := genericapifilters.Unauthorized(c.RequestContextMapper, c.Serializer, c.SupportsBasicAuth)
	if utilfeature.DefaultFeatureGate.Enabled(features.AdvancedAuditing) {
		failedHandler = genericapifilters.WithFailedAuthenticationAudit(failedHandler, c.RequestContextMapper, c.AuditBackend, c.AuditPolicyChecker)
	}
	handler = genericapifilters.WithAuthentication(handler, c.RequestContextMapper, c.Authenticator, failedHandler)
	handler = genericfilters.WithCORS(handler, c.CorsAllowedOriginList, nil, nil, nil, "true")
	handler = genericfilters.WithTimeoutForNonLongRunningRequests(handler, c.RequestContextMapper, c.LongRunningFunc, c.RequestTimeout)
	handler = genericapifilters.WithRequestInfo(handler, c.RequestInfoResolver, c.RequestContextMapper)
	handler = apirequest.WithRequestContext(handler, c.RequestContextMapper)
	handler = genericfilters.WithPanicRecovery(handler)
	return handler
}

panic recover

genericfilters.WithPanicRecovery 在 handler 的最外层对出现的 panic 恢复，并且打印每次请求的 log，所以你想观察 API 请求的情况可以 grep wrap.go 就能看到。

request context

apirequest.WithRequestContext 给 request 绑定一个 Context

RequestInfo

跟路 url 提取后续请求需要的 group, version, namespace, verb, resource 等信息。

WithTimeoutForNonLongRunningRequests

限制 API 调用时间，超时处理提前终止 write。

WithCORS

允许跨域访问。

authentication

在 k8s.io/apiserver/pkg/endpoints/filters/authentication.go 下。WithAuthentication 插入鉴权信息，例如证书鉴权，token 鉴权等，并且从鉴权信息当中获取 user 信息（可能是 service account 也可能是外部用户）user 身份是由这里面的几种方式确认的

authorization

检查是否有权限进行对应资源的操作。一种是 RBAC 一种是 Node。具体这两种方式可以看这个介绍，RBAC 主要是针对服务的，而 Node 模式主要是针对 kubelet 的。

impersonation

让用户伪装成其他用户，比如 admin 可以用普通用户的身份创建资源。

路由

通过 genericapiserver 的 InstallLegacyAPIGroup 就注册到路由当中。具体的做法就是根据 version, resource, sub resource, verb 等信息构造路由，然后用 go-restful 注册处理函数。比如说 GET

route := ws.GET(action.Path).To(handler).
	Doc(doc).
	Param(ws.QueryParameter("pretty", "If 'true', then the output is pretty printed.")).
	Operation("read"+namespaced+kind+strings.Title(subresource)+operationSuffix).
	Produces(append(storageMeta.ProducesMIMETypes(action.Verb), mediaTypes...)...).
	Returns(http.StatusOK, "OK", producedObject).
	Writes(producedObject)

handler 里面做的内容就是序列化，然后根据具体的要求（GET DELETE 等）到 etcd 中操作，当然本身还有一层缓存，这取决于 API 的 options 是希望更新还是直接读缓存（缓存会比 etcd 旧一些），比如对于 kubelet 会不断查询 node 信息，但是 kubelet 本身并不需要最新的信息，这个时候就会从缓存中读取。

性能调优

开启代理 kubectl proxy，就可以通过 localhost 直接访问 kube-apiserver HTTP 服务。然后执行 go tool pprof http://localhost:8001/debug/pprof/profile 可以获得 profile 结果，下图红色的部分就是调用耗时最多的部分。

除此之外，kube-apiserver 本身也暴露了很多 prometheus 的 metrics 但是往上现在没有现成的模板，只能根据自己的需求来在 prometheus 当作做 query。可以在 k8s.io/apiserver/pkg/endpoints/metrics/metrics.go 里面看到。

之前也说过，超时间调用时会打 log 的，在代码中保存了一些 trace 日志，可以通过 grep Trace来过滤。Trace[%d] 这样开头, %d 是一个 id 可以看到具体的 trace 信息。

kubernetes 调度器指北

发表于 2017-09-26 更新于 2025-03-14 Disqus：

最近因为 k8s 的默认调度器功能太丰富，太“高级”了，一些屌丝特性没有满足，所以前段时间自己魔改了一下满足了一些屌丝特性，暂时叫做乞丐调度器，顺便把默认的调度器代码翻了一下，这里对默认的代码做一下总结。

CreateScheduler

CreateScheduler 会从 policy file 当中获取算法的配置信息。接口k8s.io/kubernetes/plugin/pkg/scheduler.Configurator 定义了构造一个 scheduler 的配置信息。k8s.io/kubernetes/plugin/pkg/scheduler/factor.ConfigFactory 是这个接口的一个实现。c.Create 会把 AlgorithmProvider 配置的 Predicate 和 Priority 的 key 当作参数传给，f.CreateFromKeys 这样主要对应的 key 注册了的话就会有对应的算法绑定到 scheduler 上面。目前有两种 Provider，一种是用默认 predicate 和默认 priority 的 AlgorithmProvider，另一种是把 LeastRequestedPriority 换成 MostRequestedPriority 的自动伸缩友好的 AlgorithmProvider。

// Registers algorithm providers. By default we use 'DefaultProvider', but user can specify one to be used
// by specifying flag.
factory.RegisterAlgorithmProvider(factory.DefaultProvider, defaultPredicates(), defaultPriorities())
// Cluster autoscaler friendly scheduling algorithm.
factory.RegisterAlgorithmProvider(ClusterAutoscalerProvider, defaultPredicates(),
	copyAndReplace(defaultPriorities(), "LeastRequestedPriority", "MostRequestedPriority"))

Scheduler Server

options.ScheduleServer 是服务端对应的配置结构，其中有几个成员。

KubeSchedulerConfiguration 调度器的配置
Master 表示 API server 的地址
Kubeconfig k8s 配置文件的路径

func Run(s *options.SchedulerServer) error 会根据 Sechduler Server 来运行。EventBroadcaster 接受事件，并且把事件发送给事件处理者（ EventSink watcher, log)，startHTTP 主要是是 profiling 接口，心跳检测接口和 prometheus 的 instrumenting 接口。informerFactory，看起来是一个异步同步信息的 cache，平时调度是直接走 cache，更新的时候才会走 API。最后配置了选主的话会从 Etcd 拿到锁，并且拿到 Master 的锁。

k8s.io/kubernetes/plugin/pkg/scheduler.Scheduler.Run

初始化以后，Run 对应的是一个 0 秒循环的大 loop（相当于每次 loop 等于主动调用一次 Go runtime.Sched())，在每次循环当中都会调用 sched.scheduleOne，首先 NextPod 会同步等待一个 pod 变成 available 的状态，并且跳过正在被删除的 pod，然后调用 sched.schedule 走到具体的调度算法当中，整个过程是串行，没有批量调度 pod 的操作。在进行具体的调度算法之后，会得到一个可行的 node，如果调度失败的话会，并且调度失败的原因是找不到合适的 node 的话，就尝试 sched.preempt，这个的作用就是尝试在替换现有 pod 的情况下能够获得调度机会的策略，那么就抢占已经被调度的 pod，标记目标 pod 的 Annotation 然后踢出权重最低的那个 pod。如果成果获得一个可调度的节点，就通过把本地 cache 先更新到已经调度之后的状态，标记 pod 已经在要调度的 node 上，也就是调用 sched.assume 假设 pod 已经调度到了节点上，再异步的通过 ApiServer 的接口，sched.bind 让 pod 正在运行到 node 上。

sched.schedule

schedule 调用 algorithm/scheduler_interface.go下面定义的调度器的接口 Schedule。对应的实现在 core/generic_scheduler.go 下面。

// ScheduleAlgorithm is an interface implemented by things that know how to schedule pods
// onto machines.
type ScheduleAlgorithm interface {
	Schedule(*v1.Pod, NodeLister) (selectedMachine string, err error)
	// Preempt receives scheduling errors for a pod and tries to create room for
	// the pod by preempting lower priority pods if possible.
	// It returns the node where preemption happened, a list of preempted pods, and error if any.
	Preempt(*v1.Pod, NodeLister, error) (selectedNode *v1.Node, preemptedPods []*v1.Pod, err error)
	// Predicates() returns a pointer to a map of predicate functions. This is
	// exposed for testing.
	Predicates() map[string]FitPredicate
	// Prioritizers returns a slice of priority config. This is exposed for
	// testing.
	Prioritizers() []PriorityConfig
}

Schedule 会根据调度算法得到一个合适的节点，而 Preempt 则是尝试抢占一个 pod 以获得调度到节点上的机会。Predicates 和 Priorities 则是两个重要的部分，Predicates 类似一个过滤器，对节点进行筛选，而 Priorities 则是对筛选出来的节点进行权重的排序，最后得到一个合适的调度节点。

算法工厂

算法工厂就是注册 Preciate 和 Priority 的地方，之前已经说了可以通过 AlgorithmProvider 获得一组 Predicate 和 Priority，比如 DefaultProvider 提供了默认的一套，如果不用 Provider，需要在 policy file 当中另外指定要使用的 Predicate 和 Priority，不过目前好像没有用这种方式，还是通过 Provider 指定了一套要使用的算法。注册 AlgorithmProvider 是通过factory.RegisterAlgorithmProvider，然后调用 NewGenericScheduler (在 k8s.io/kubernetes/plugin/pkg/scheduler/core/generic_scheduler.go 当中)，初始化要用到的 predicates 和 priorities。

调度过程

Schedule 其实很简单，就是通过 findNodesThatFit，先根据 Predicate 过滤出合适的 Node，然后调用 PrioritizeNodes，用 Priorities 对 Node 根据算法的权重进行排序，因为每个 node 要走的流程是一样的并且最终结果相互没有影响，所以这个过程是并发的，这篇文章的图画的很好，示意很明显。

最后会得到一个最理想的节点，再通过 bing 告诉 API server 这个节点被选中了。

sched.preempt

抢占过程是在 pod 没有找到合适的节点情况下，如果能在踢出一个 pod 获得调度机会的情况下进行抢占。抢占算是一个比较新的特性，在 1.8 里面都是默认关掉的，要打开的话需要指定kube-scheduler 的 --feature-gates=PodPriority=true 还有 apiserver 的 --runtime-config=scheduling.k8s.io/v1alpha1=true。可以通过添加 PriorityClass 把 pod 分权重，现在这个特性算是给 pod 也加上的权重。

apiVersion: v1
kind: PriorityClass
metadata:
  name: high-priority
value: 1000000
globalDefault: false
description: "This priority class should be used for XYZ service pods only."

然后可以在spec当中指定 priorityClassName: high-priority，这样这么大的权重，这个 pod 就很难被抢占了，具体流程如下图。

podEligibleToPreemptOthers 主要判断如果目标 pod 被标记为（通过 pod 的 annotation 标记）已经要抢占其他 pod，并且有一个优先级小于目标 pod 的 pod 即将被删除 (p.DeletionTimestamp != nil )，就直接退出，因为这个时候这个被删除的 pod 其实在为目标 pod 腾出空间了，在下次调度的时候就会获得调度机会。nodesWherePreemptionMightHelp，类似于 schedule 的时候的 predicate 阶段，只不过多了一步是通过尝试移除 pod 跑一遍 predicates 看看这个节点能不能被通过。 selectNodesForPreemption 则和 priority 的阶段类似，把删除 pod 之后的可以通过的节点进行排序选出一个排名最高的节点。再通过 selectVictimsOnNode 把节点上的 pod 按照节点的 priority 排序选出“受害者”，越高越难被抢占。可以在 spec 里面设置这个值，选出了节点上的受害者以后，通过pickOneNodeForPreemption，主要的依据是拥有最低的最高 pod 权重的节点先被选出来，比如 node1 上 pod 的最高权重是 10，node2 上 pod 的最高权重是 8，那么 node2 被选中，如果有平局的话，最少的“受害者”先选，如果还平局，随机选一个。最后得到一个要被抢占的节点。

自定义调度器的方式

自定义调度器有三种方法。

第一种是通过添加 Predicate 和 Priority 的方式，做微调，这种方式比较简单，只要定义好对应的函数并且通过函数工厂注册就可以。

第二种是使用自定义的调度器，具体的方法可以看官方文档，通过把 pod 的 spec.schedulerName 指向自定义的调度器就可以把调度任务转到自己实现的服务。

第三种是使用 extender，extender 本身和调度器的过程类似，接口是如下定义的，主要是针对一些不算受集群本身控制的资源，需要通过外部调用来进行调度的情况，相关文档在这里。

性能测试

目前单机简单的测试条件下，1s 钟可以调度成功 450 左右的 pod，具体的性能参数还要慢慢挖掘。

flannel 网络架构

发表于 2017-09-21 更新于 2025-03-14 Disqus：

flannel 是一个中心化的 overlay 容器网络，设计简单，容易理解，对于 k8s 来说，有一个假设：所有容器都可以和集群里任意其他容器或者节点通信，并且通信双方看到的对方的 IP 地址就是实际的地址，主要的好处就是不需要任何的端口映射和地址转换，拥有一张扁平的网络更容易管理，而且由于是基于 Etcd 的中心化的管理，所以对于一些 IP 变化异常频繁的场景来说，比一些去中心化的方案能够较及时同步网络拓扑关系。

IP 地址的管理

flannel 的 IP 地址是通过 Etcd 管理的，在 k8s 初始化的时候指定 pod 大网的网段 --pod-network-cidr=10.244.0.0/16，flanneld 可以直接通过 Etcd 管理，如果启动的时候指定了 --kube-subnet-mgr，可以直接通过 k8s 的 apiserver 来获得一个小网段的租期，通过 kubectl get <NodeName> -o jsonpath='{.spec.podCIDR}' 可以获取对应节点的 CIDR 表示的网段，flannel 是以节点为单元划分小网段的，每个节点上的 pod 在这个例子当中是划分一个 10.244.x.0/24 的网段，所以总共能分配 255 个节点，每个节点上可以分配 253 个 pod。结构如下图所示，每个节点上都会有一个 flanneld 用于管理自己网段的租期。

可以通过在 host 上 cat /run/flannel/subnet.env 查看同步下来的信息，例如：

FLANNEL_NETWORK=10.244.0.0/16
FLANNEL_SUBNET=10.244.0.1/24
FLANNEL_MTU=8951
FLANNEL_IPMASQ=true

说明当前节点分配的网段是 10.244.0.1/24。在每个节点上因为已经确定了网段，用 ipam 就可以管理这一范围 ip 地址的分配，所以本身 pod 的 IP 分配和中心 Etcd 没有太多联系。

基本工作原理

简单来说就是通过建立 VXLAN 隧道，通过 UDP 把 IP 封装一层直接送到对应的节点，实现了一个大的 VLAN。没有使用 IPoIP 或者 GRE 主要是因为一些云厂商比如 AWS 的安全策略只能支持 TCP/UDP/ICMP。

flannel 本身会创建一个类似下面这样配置的 CNI bridge 设备。

{
    "name" : "cni0",
    "type" : "bridge",
    "mtu" : 8973,
    "ipMasq" : true,
    "isGateway" : true,
    "ipam" : {
        "type" : "host-local",
        "subnet" : "10.244.0.1/24",
         "routes" : [ { "dst" : "10.244.0.0/16" } ]
    }
}

具体的网络拓扑图如下，所用的网段开头是 10.1，但是划分是一致的，图里面的 docker0 应该是 cni0，flannel0 应该是 flannel.1，这个命名的区别主要是不带点的是 UDP 封装，带点的是 vxlan 封装，图片比较早。

第一步，首先是从容器中(10.1.20.3)出来，走桥接去到 cni0 (10.1.20.1)，通过brctl show 可以看到 cni0 接了两个容器的 veth。

第二步，然后根据路由规则，在宿主机上能够用ip route，找到一条走到大网段的路由。10.15.0.0/16 dev flannel.1，到达 flannel.1。

第三步，flannel.1 会走 vxlan，这个是在内核实现的，如果用 UDP 封装就是在用户态实现的，用户态实现的等于把包从内核过了两遍，没有直接用 vxlan 封装的直接走内核效率高，所以基本上不会使用 UDP 封装。对应的 vxlan 配置可以通过 bridge fdb flannel.1 看到，没有一条隧道就会有一条这样的转发表。因为到达每个对应网段的信息是在 Etcd 上分配的 flannel.1 只要 watch 然后发现有更改的时候对应配置隧道指向对应容器网段的宿主机 IP 就可以。

1	8a:55:a7:e2:e9:18 dev flannel.1 dst 192.168.0.100 self permanent

第四步，走宿主机的网络到达对端，对端的 vxlan 收到以后会拆开封装，丢到协议栈里面。

第五步，根据路由 ip route，中的一条10.1.15.0/24 dev cni0 proto kernel scope link src 10.1.15.1，送到 cni0 然后再转发给容器 10.1.15.2, 最后就能完成 pod 跨界点的互通了。

优缺点

因为整个的网段分配是存在 Etcd 里面的，节点只要 watch 然后根据网段建隧道就可以，相对来说中心化的系统设计比较简单，而且对于 IP 地址变动能够及时反应，特别是节点和容器都有剧烈变化的时候（别问我为啥物理节点会有剧烈变化，创业公司玩法怎么省钱怎么来……），相比于去中心化的一些设计能够更快同步一些。当然建隧道是一个点对点的规模，也就是如果有 n 个节点建隧道的话，每个节点上都要建 n-1 条隧道。

一种改进方式是使用 host-gw 的后端方式，以及 ipvlan，不过目前 ipvlan 还没有支持，这里有一个各种后端实现的比较，vxlan 表现很最差，host-gw 的做法是不用隧道的方式，而是把路由信息直接写到节点上，直接打平到节点上，等于是节点之间是一个大网，每个节点上的小网段通过路由的方式和大网互通，将到达各个节点的网段的路由刷到节点上，而不是建 vxlan 隧道的方式，比如文中的例子，会有这样的路由。

1 2	10.1.15.0/24 via 192.168.0.100 dev eth0 10.1.15.0/24 dev cni0 proto kernel scope link src 10.1.20.1

然而，由于 flannel 只能够修改各个主机的路由表，一旦主机直接隔了个其他路由设备，比如三层路由器，这个包就会在路由设备上被丢掉。这样一来，host-gw 的模式就只能用于二层直接可达的网络。

LC-trie 快速路由查找算法

发表于 2017-09-05 更新于 2025-03-14 Disqus：

一般来说我们把字典树称作 trie，这是一个用于查找字符串的数据结构，举个简单的例子。下面就是一个把 b，abc，abd，bcd，abcd，efg，hii 这 6 个单词构造成字典树的例子。

在路由表中我们可以把子网看作一个二进制字符串，对于 IPv4 来说，字符串的长度就有 32 位。

以上面这个为例子，有编号 0 到 14 这几个二进制的字符串（字符集和只有 0 和 1），右边是他们的具体字符串，这样的字典树有一点像二叉树。如果把它画出来是这样的。

对这个树做压缩有一个办法就是把单节点作压缩，形成下面这个树，这个就是路径压缩，path compressed trie。

我们需要在压缩的地方进行标记，比如 skip 4 那里把一段 1011 压缩掉了。而 LC-trie 指的是 Level Compressed trie，这个树会把层级进行压缩，因为我们得到的这个字典树实际上是一个稀疏树，高度并不是平衡的，所以为了达到平衡，需要做的一件事情是把高度进行压缩，压缩之后变成下面这个形式，这样整棵树就会更加扁平。

这个树是有两种类型的节点，一种是 leaf，保存了路由具体的信息的叶子结点，一种是 trie node（tnode）保存了中间节点，子节点可能是 tnode 或者 leaf。trie 上有几个数据要存储，一个是 bits，这个表示的是子节点的选择度（这个怎么理解呢，就是我接下来的子节点是八叉了，因为我把原来的树压缩了，所以现在不需要二选一，现在直接来个八选一就可以），对于一个八叉的压缩来说，就要有一个 3 位的数字来存储，也就是 log(8)，当然也可以能是 7 叉，但是必须是 2 的指数。而 pos 表示的是从那个 bit 开始，它的作用和 skip 类似只不过不是一个相对值，而是一个累加值，表示我们要从哪开始（从字符串的起始位置开始数）。

我们先看一下搜索的代码，其实很简单，就是不断匹配公共前缀直到直到找到叶子节点（ bits == 0）。匹配前缀的方式比较 tricky，用异或进行确认。

/* rcu_read_lock needs to be hold by caller from readside */
static struct key_vector *fib_find_node(struct trie *t,
					struct key_vector **tp, u32 key)
{
	struct key_vector *pn, *n = t->kv;
	unsigned long index = 0;

	do {
		pn = n;
		n = get_child_rcu(n, index);

		if (!n)
			break;

		index = get_cindex(key, n);

		/* This bit of code is a bit tricky but it combines multiple
		 * checks into a single check.  The prefix consists of the
		 * prefix plus zeros for the bits in the cindex. The index
		 * is the difference between the key and this value.  From
		 * this we can actually derive several pieces of data.
		 *   if (index >= (1ul << bits))
		 *     we have a mismatch in skip bits and failed
		 *   else
		 *     we know the value is cindex
		 *
		 * This check is safe even if bits == KEYLENGTH due to the
		 * fact that we can only allocate a node with 32 bits if a
		 * long is greater than 32 bits.
		 */
		if (index >= (1ul << n->bits)) {
			n = NULL;
			break;
		}

		/* keep searching until we find a perfect match leaf or NULL */
	} while (IS_TNODE(n));

	*tp = pn;

	return n;
}

具体看一下 get_cindex 匹配的方式，#define get_cindex(key, kv) (((key) ^ (kv)->key) >> (kv)->pos)，对 tnode 和被比较的 key 做异或，这个怎么理解呢，看下面的图的例子这是正确匹配以后的结果，灰色代表 0，蓝色代表 1，两个值进行异或的话，首先 pos 会被右移掉，然后 bits 的部分会原样保留，因为 tnode 的这部分都是 0。然后亲址的部分如果完全匹配的话结果就都是 0 ，但是如果不完全匹配的话，结果就会比 index 还要大，因为高位还有 1，所以这就是为什么 index >= (1ul << n->bits 能判断是否匹配的前缀的原因。

然后我们再看一下插入的流程。首先算出匹配到当前节点的子节点（有可能有，有可能没有）。


static int fib_insert_node(struct trie *t, struct key_vector *tp,
			   struct fib_alias *new, t_key key)
{
	struct key_vector *n, *l;

	l = leaf_new(key, new);
	if (!l)
		goto noleaf;

	/* retrieve child from parent node */
	n = get_child(tp, get_index(key, tp));

如果有子节点，就要创建一个新的 tnode，再把这个 key 给插入。

/* Case 2: n is a LEAF or a TNODE and the key doesn't match.
 *
 *  Add a new tnode here
 *  first tnode need some special handling
 *  leaves us in position for handling as case 3
 */
if (n) {
	struct key_vector *tn;

__fls find last set bit，就是找到 pos，然后扩展出有两个选择（2 的 1 次方）的 tnode。

1
2
3

tn = tnode_new(key, __fls(key ^ n->key), 1);
if (!tn)
	goto notnode;

设置 tn 的父节点为 tp，然后把 key 插入到 tn 当中。

	/* initialize routes out of node */
	NODE_INIT_PARENT(tn, tp);
	put_child(tn, get_index(key, tn) ^ 1, n);

	/* start adding routes into the node */
	put_child_root(tp, key, tn);
	node_set_parent(n, tn);

	/* parent now has a NULL spot where the leaf can go */
	tp = tn;
}

/* Case 3: n is NULL, and will just insert a new leaf */
node_push_suffix(tp, new->fa_slen);
NODE_INIT_PARENT(l, tp);
put_child_root(tp, key, l);

开始进行平衡调整树形。

	trie_rebalance(t, tp);

	return 0;
notnode:
	node_free(l);
noleaf:
	return -ENOMEM;
}

再看一下树的高度是如何进行调整的，从当前节点一直向上压缩。

static void trie_rebalance(struct trie *t, struct key_vector *tn)
{
	while (!IS_TRIE(tn))
		tn = resize(t, tn);
}

#define MAX_WORK 10
static struct key_vector *resize(struct trie *t, struct key_vector *tn)
{
#ifdef CONFIG_IP_FIB_TRIE_STATS
	struct trie_use_stats __percpu *stats = t->stats;
#endif

利用 container_of 获取父节点。

1	struct key_vector *tp = node_parent(tn);

获取子节点，初始化 max_work 为 10。

unsigned long cindex = get_index(tn->key, tp);
int max_work = MAX_WORK;

pr_debug("In tnode_resize %p inflate_threshold=%d threshold=%d\n",
	 tn, inflate_threshold, halve_threshold);

/* track the tnode via the pointer from the parent instead of
 * doing it ourselves.  This way we can let RCU fully do its
 * thing without us interfering
 */
BUG_ON(tn != get_child(tp, cindex));

/* Double as long as the resulting node has a number of
 * nonempty nodes that are above the threshold.
 */

should_inflate 决定要不要压缩的依据是根据动态压缩算法来的（引用4），直观的来说就是高度超过了一个动态计算的阈值，并且还没压缩超过十次就会继续压缩。这个动态阈值的算法是用非空子节点的数目如果超过压缩之后子节点数目的一半就值得压缩。而 inflate 做的事情就把层级压缩一层，也就是把 children 的 children 按照 bits 的匹配放到 parent 的 new_children 当中。

	while (should_inflate(tp, tn) && max_work) {
		tp = inflate(t, tn);
		if (!tp) {
#ifdef CONFIG_IP_FIB_TRIE_STATS
			this_cpu_inc(stats->resize_node_skipped);
#endif
			break;
		}

		max_work--;
		tn = get_child(tp, cindex);
	}

	/* update parent in case inflate failed */
	tp = node_parent(tn);

	/* Return if at least one inflate is run */
	if (max_work != MAX_WORK)
		return tp;

到这里说明一次调整都没有发生，说明节点很稀疏，也就是把节点分开。

	/* Halve as long as the number of empty children in this
	 * node is above threshold.
	 */
	while (should_halve(tp, tn) && max_work) {
		tp = halve(t, tn);
		if (!tp) {
#ifdef CONFIG_IP_FIB_TRIE_STATS
			this_cpu_inc(stats->resize_node_skipped);
#endif
			break;
		}

		max_work--;
		tn = get_child(tp, cindex);
	}

只有一个孩子，可以进行 path compress，没必要再多一个中间节点。

	/* Only one child remains */
	if (should_collapse(tn))
		return collapse(t, tn);

	/* update parent in case halve failed */
	return node_parent(tn);
}

整个的 LC-trie 的结构大致如此，主要用于路由表中路由规则的快速匹配，是在 3.6 之后引进的，摒弃了之前用哈希来查找路由表的算法。

引用

neighboring subsystem 浅析

发表于 2017-08-07 更新于 2025-03-14 Disqus：

neighbor 在协议栈里指的是同一个 LAN 下面的邻居，也就是他们在 L3 上通过媒介或者点对点连接在了一起。邻居子系统同时也可以理解为一个 L2 和 L3 地址的转换器。邻居子系统的目的就在于上层协议不应该关心下层的地址信息，我发送过来的 IP 地址应该让下层来决定发送到哪个 MAC 地址。neighbor solicitation and neighbor advertisement，可以分别对应 ARP 的 request 请求和 ARP 的 reply 请求。

邻居子系统主要缓存两大块的内容，一个是 L3 到 L2 的映射解析的缓存，一个是 L2 头的缓存，缓存 L2 头的原因是大部分 L2 的头基本上是重复的，所以通过缓存头部可以加快协议的封装。

以下有几个相关结构体需要介绍一下。

struct neighbour 代表的是一个邻居的信息，比如 L2 和 L3 地址等。

struct neigh_table 代表的是一种邻居协议的接口(比如 ARP)。

struct neigh_params 代表的是邻居协议在每个设备上的不同参数。

struct neigh_ops 邻居对应的一些操作函数。

struct hh_cache 缓存 L2 的头部，不是所有的设备都支持头部缓存。

struct rttable 和 struct dst_enry， IPv4 的路由缓存信息是通过 struct rttable 缓存的。

这是 dst_entry, hh_cache 和 neighbour 之间的关系。

neighbor 结构是用 hash 存储的，key 是 L3 地址加设备（对应的 device 结构体）加一个随机值。hash 表通过 neigh_hash_alloc 和 neigh_hash_free 来分配和释放。neigh_lookup 就是通过 key 从 hash table 中找到对应的 neighbour 结构体，实现都不复杂，这里理解就可以了。

一般邻居子系统的缓存流程是这样的，如果 L3 的请求到达，地址解析缓存没有命中，把 L3 packet 入队，开始neighbor solicitation 等收到 neighbor advertisement 之后再出队并且发送。

邻居状态信息 NUD(Netowork Unreachablility) 状态

NUD state 本来是 IPv6 协议里的邻居关系的定义，但是在内核中沿用到了 IPv4 里面。

NUD_NONE 这个 neighbour 刚开始建立，没有相关状态
NUD_INCOMPLETE 发送了获取 L2 地址的请求（通过 ARP 或者其他协议），但是没收到回复，并且之前不存在老缓存
NUD_REACHABLE neighbour 地址被缓存了，并且是可达的
NUD_FAILED 因为发送地址解析请求失败了，标志邻居不可达
NUD_STALE NUD_DELAY NUD_PROBE 是解析请求确认邻居可达的过程中的中间状态
NUD_NOARP 标志不需要用解析协议（虽然是 NOARP，但是别的协议也用这个标记），这是一个特殊情况。
NUD_PERMANET 标志邻居的地址解析是静态配置的

这些是基本状态，根据这些基本状态组合了几个相对有语义的状态。

NUD_VALID 表示该地址会是一个有效的地址

NUD_PERMANENT NUD_NOARP NUD_REACHABLE NUD_PROBE NUD_STALE NUD_DELAY
NUD_CONNECTED 是 NUD_VALID 的子集，去除了待决的中间状态

NUD_PERMANENT NUD_NOARP NUD_REACHABLE
NUD_IN_TIMER 表示在这个状态下正在执行一个定时任务，一般是状态不明了的时候

NUD_INCOMPLETE NUD_DELAY NUD_PROBE

邻居的可达性可以通过两点来确认，收到了一个地址解析协议的单播回复，或者通过外部信息确认（比如收到了这个邻居的 TCP 报文，当然这个 IP 可能不是自己的邻居，但是可以可以确定对应网关『作为邻居』的可达性）。

确认邻居信息

IP 层会调用ipv4_confirm_neigh 来确认映射地址，如果有 gateway 用 gateway，没有 gateway 开始查缓存。在 net/ipv4/ip_output.c 中的对应代码，就是查缓存的 neighbour 结构体，如果这个结构体不存在的话就要开始confirm 了。

neigh = __ipv4_neigh_lookup_noref(dev, nexthop);

if (unlikely(!neigh))
	neigh = __neigh_create(&arp_tbl, &nexthop, dev, false);
if (!IS_ERR(neigh)) {
	int res;

	sock_confirm_neigh(skb, neigh);
	res = neigh_output(neigh, skb);

	rcu_read_unlock_bh();
	return res;
}

当 TCP 收到报文（比如对端的 SYN/ACK 时），这种外部信息说明其实这个节点是可达的（不是来自 gateway），也可以更新缓存。

另外，neigh_connect 和 neigh_suspect 是两个状态转换时会调用的函数。

当 neigh 进入 NUD_REACHABLE ， neigh_connect 把 neigh->output 的函数指向 connected_output 这个函数，它会在调用 dev_queue_xmit 之前填充 L2 头部，把包直接发出去。

当从 NUD_REACHBLE 转换成 NUD_STALE 或者 NUD_DELAY，neigh_suspect 会强制进行可达性的确认，通过把 neighbor->output 指向 neigh_ops->output，也就是 neigh_resolve_output，它会在调用 dev_queue_xmit 之前先把地址解析出来，等把地址解析完成以后再把缓存的包发送出去。

更新邻居信息

邻居信息更新的入口函数就是 neigh_update，这个函数定义如下。

1 2	int neigh_update(struct neighbour neigh, const u8 lladdr, u8 new, u32 flags, u32 nlmsg_pid)

首先做一些预先检查，如果是管理员配置，原来的配置就不能是 PERMANENT 或者 NOARP 的。

1
2
3

if (!(flags & NEIGH_UPDATE_F_ADMIN) &&
    (old & (NUD_NOARP | NUD_PERMANENT)))
	goto out;

标记为 dead，然后 goto out

1 2	if (neigh->dead) goto out;

如果更新为无效的标记的话，删除 timer，并且 supect 一下，如果是 NUD_CONNECTED 状态。
如果是需要标记为失败的 neigh(之前是 INCOMPLETE|NUD_PROBE)，则调用 neigh_invalidate，让这个 neigh 无效。

if (!(new & NUD_VALID)) {
	neigh_del_timer(neigh);
	if (old & NUD_CONNECTED)
		neigh_suspect(neigh);
	neigh->nud_state = new;
	err = 0;
	notify = old & NUD_VALID;
	if ((old & (NUD_INCOMPLETE | NUD_PROBE)) &&
	    (new & NUD_FAILED)) {
		neigh_invalidate(neigh);
		notify = 1;
	}
	goto out;
}

接下来是三个条件，一个是 device 没有硬件地址，用 neigh->ha，如果 lladdr 提供了并且老的是有效的，使用老的地址，如果 lladdr 没有提供，直接使用老的地址。

/* Compare new lladdr with cached one */
if (!dev->addr_len) {
	/* First case: device needs no address. */
	lladdr = neigh->ha;
} else if (lladdr) {
	/* The second case: if something is already cached
	   and a new address is proposed:
	   - compare new & old
	   - if they are different, check override flag
	 */
	if ((old & NUD_VALID) &&
	    !memcmp(lladdr, neigh->ha, dev->addr_len))
		lladdr = neigh->ha;
} else {
	/* No address is supplied; if we know something,
	   use it, otherwise discard the request.
	 */
	err = -EINVAL;
	if (!(old & NUD_VALID))
		goto out;
	lladdr = neigh->ha;
}

如果NUD_CONNECTED 更新 confirm 的时间，更新『更新』的时间。

1
2
3

if (new & NUD_CONNECTED)
	neigh->confirmed = jiffies;
neigh->updated = jiffies;

NEIGH_UPDATE_F_OVERRIDE_ISROUTER 标记的是当前 neigh 是一个 router。

/* If entry was valid and address is not changed,
   do not change entry state, if new one is STALE.
 */
err = 0;
update_isrouter = flags & NEIGH_UPDATE_F_OVERRIDE_ISROUTER;
if (old & NUD_VALID) {
	if (lladdr != neigh->ha && !(flags & NEIGH_UPDATE_F_OVERRIDE)) {
		update_isrouter = 0;
		if ((flags & NEIGH_UPDATE_F_WEAK_OVERRIDE) &&
		    (old & NUD_CONNECTED)) {
			lladdr = neigh->ha;
			new = NUD_STALE;
		} else
			goto out;
	} else {
		if (lladdr == neigh->ha && new == NUD_STALE &&
		    !(flags & NEIGH_UPDATE_F_ADMIN))
			new = old;
	}
}

如果是更新操作，删除老的 timer，如果需要 timer，更新新的 timer，并且设置新状态。

if (new != old) {
	neigh_del_timer(neigh);
	if (new & NUD_PROBE)
		atomic_set(&neigh->probes, 0);
	if (new & NUD_IN_TIMER)
		neigh_add_timer(neigh, (jiffies +
					((new & NUD_REACHABLE) ?
					 neigh->parms->reachable_time :
					 0)));
	neigh->nud_state = new;
	notify = 1;
}

更新 neigh->ha，如果 lladdr 和 neigh->ha 不同的话。

if (lladdr != neigh->ha) {
	write_seqlock(&neigh->ha_lock);
	memcpy(&neigh->ha, lladdr, dev->addr_len);
	write_sequnlock(&neigh->ha_lock);
	neigh_update_hhs(neigh);
	if (!(new & NUD_CONNECTED))
		neigh->confirmed = jiffies -
			      (NEIGH_VAR(neigh->parms, BASE_REACHABLE_TIME) << 1);
	notify = 1;
}
if (new == old)
	goto out;

根据状态调用 connect 和 suspect

if (new & NUD_CONNECTED)
	neigh_connect(neigh);
else
	neigh_suspect(neigh);

如果之前老的不是 NUD_VALID，就会把 skb 从 arp_queue，并且释放 arp_queue。

if (!(old & NUD_VALID)) {
	struct sk_buff *skb;

	/* Again: avoid dead loop if something went wrong */

	while (neigh->nud_state & NUD_VALID &&
	       (skb = __skb_dequeue(&neigh->arp_queue)) != NULL) {
		struct dst_entry *dst = skb_dst(skb);
		struct neighbour *n2, *n1 = neigh;
		write_unlock_bh(&neigh->lock);

		rcu_read_lock();

		/* Why not just use 'neigh' as-is?  The problem is that
		 * things such as shaper, eql, and sch_teql can end up
		 * using alternative, different, neigh objects to output
		 * the packet in the output path.  So what we need to do
		 * here is re-lookup the top-level neigh in the path so
		 * we can reinject the packet there.
		 */
		n2 = NULL;
		if (dst) {
			n2 = dst_neigh_lookup_skb(dst, skb);
			if (n2)
				n1 = n2;
		}
		n1->output(n1, skb);
		if (n2)
			neigh_release(n2);
		rcu_read_unlock();

		write_lock_bh(&neigh->lock);
	}
	__skb_queue_purge(&neigh->arp_queue);
	neigh->arp_queue_len_bytes = 0;
}

更新 flag，发送 notify 通过 rtnetlink 通知 nlmsg_pid 对应的进程，

out:
if (update_isrouter) {
	neigh->flags = (flags & NEIGH_UPDATE_F_ISROUTER) ?
		(neigh->flags | NTF_ROUTER) :
		(neigh->flags & ~NTF_ROUTER);
}
write_unlock_bh(&neigh->lock);

if (notify)
	neigh_update_notify(neigh, nlmsg_pid);

这就是邻居信息的更新流程。

当然还有一个缓存项是 L2 header 的缓存这里就简单略过了。

ARP

接下来说一下 ARP 相关的内容，ARP 本身的格式其实很简单。

邻居子系统有一个比较关键的协议就是 ARP，这里简单介绍一下 ARP 协议，我们来看一下 ARP 的格式。

先是两个字节的 L2 类型，然后是两个字节的 L3 类型。一般 L2 是 1 (以太网)，L3 是 0x0800 (IPV4)，接着跟着的是 L2 类型的长度和 L3 类型的长度，这里 Ethernet 的 MAC 地址是 6 个字节，IPV4 的 IP 地址是 4 个字节，接着是两个字节的操作，1 表示请求，2 表示回复，然后是发送端 L2 的地址，L3 的地址，接着是接收端 L2 和 L3 的地址。整个翻译过来就是『我（MAC 是 0a:00:27:00:00:00 IP 是 192.168.56.1）广而告之（MAC 广播地址），问一下谁知道 192.168.56.102 的地址』。

gARP (gratuitous ARP)

听起来像是一个无理由的 ARP 请求，实际上是一种主动通知的 ARP 请求。gARP 本身不是一个查询请求，而是一个通知请求，主要运用于主动通知 L2 地址改变，重复地址发现（请求解析自己的地址，如果收到回复说明有地址重复）。

还有就是 VIP，一般的作用是在本地网路中有两台机器，一台作为备机，一台作为主机，当主机 failover 的时候，备机可以继续『冒充』主机的 IP 地址，具体的做法就是主动发送请求，解析的 MAC 和 IP 都和 source 一样，老的 server 肯定不会回答这个 ARP，交换机上已经没有这个端口的缓存，会进行广播，让所有的接收者都会更新自己的缓存。也就是发送了一个一去不复返的请求，让所有的邻居更新了自己的 ARP 缓存，从而替代了老 server 的 IP，这就是 VIP 通过 ARP 实现的 failover。

总结

邻居子系统很大一部分的作用就是解析和缓存地址映射，主要是通过 ARP 来完成，而且 ARP 本身也有很多使用的姿势，也就上面说到的 gARP，邻居子系统是沟通路由子系统，以及 L2 和 L3 的桥梁。

参考：

Understanding Linux Network Internals

IP 的分片与重组

发表于 2017-07-10 更新于 2025-03-14 Disqus：

ip 分片的主体函数在 ip_fragment 当中，重组则在 ip_defrag 当中。第一个分片的标志 Offset 为 0，MF 为 1，之后的分片则是 Offset 非 0，MF 为 1，最后一个分片则是 Offset 非 0，但是 MF 为 0。以此来分别当前的 IP packet 是否是一个分片。从 IP 层向上层协议发送数据包的时候就会进行重组，比如在 ip_local_deliver 当中，调用了。说一句题外话， TCP 有 MSS ，保障 TCP message 不超过分片大小，这样是一种对底层协议有感知的行为。

if (ip_is_fragment(ip_hdr(skb))) {
	if (ip_defrag(net, skb, IP_DEFRAG_LOCAL_DELIVER))
		return 0;
}

IP 分片

ip_is_fragment 对应的条件就是 (iph->frag_off & htons(IP_MF | IP_OFFSET)) != 0;。

在 ip_fragment 当中会碰到几种情况，一种是不需要分片的 IP packet，这种很好，省心，一种是需要分片的 IP packet，这种最操心，还有一种是已经按分片负载的长度分配好了 buffer 只要加个头就相当于分片完成了就也非常棒。要从头开始进行分配的情况属于慢速路径，而已经有 buffer 准好的，直接加个头就完事的属于快速路径，快速路径的内存拷贝代价更低。

ip_fragment 主要检查 IP 是否允许进行分片，不然的话就返回一个 ICMP 错误。

struct iphdr *iph = ip_hdr(skb);

if ((iph->frag_off & htons(IP_DF)) == 0)
	return ip_do_fragment(net, sk, skb, output);

if (unlikely(!skb->ignore_df ||
	     (IPCB(skb)->frag_max_size &&
	      IPCB(skb)->frag_max_size > mtu))) {
	IP_INC_STATS(net, IPSTATS_MIB_FRAGFAILS);
	icmp_send(skb, ICMP_DEST_UNREACH, ICMP_FRAG_NEEDED,
		  htonl(mtu));
	kfree_skb(skb);
	return -EMSGSIZE;
}

return ip_do_fragment(net, sk, skb, output);

然后进入到 ip_do_fragment 当中。我们先看一下慢速路径是如何处理。

首先知道 IP 头部的长度，已经负载 (left)，然后当前的指针，已经链路层需要预留的长度。

slow_path:
	iph = ip_hdr(skb);

	left = skb->len - hlen;		/* Space per frame */
	ptr = hlen;		/* Where to start from */

	ll_rs = LL_RESERVED_SPACE(rt->dst.dev);

IP 的 offset，以及不是最后一个分片的标志位，这里是进行分片的，不知道为什么要获取一些重组时候需要的数据，TODO。

/*
 *    Fragment the datagram.
 */

offset = (ntohs(iph->frag_off) & IP_OFFSET) << 3;
not_last_frag = iph->frag_off & htons(IP_MF);

调整要分配的 skb_buff 的长度，首先不能超过 mtu，然后最后一段要按 8 对齐。

/*
 *	Keep copying data until we run out.
 */

while (left > 0) {
	len = left;
	/* IF: it doesn't fit, use 'mtu' - the data space left */
	if (len > mtu)
		len = mtu;
	/* IF: we are not sending up to and including the packet end
	   then align the next start on an eight byte boundary */
	if (len < left)	{
		len &= ~7;
	}

	/* Allocate buffer */
	skb2 = alloc_skb(len + hlen + ll_rs, GFP_ATOMIC);
	if (!skb2) {
		err = -ENOMEM;
		goto fail;
	}

设置分片的元数据，ip_copy_metadata 会拷贝优先级，协议类型，等辅助信息。然后保留 L2 的头部空间，接着在保留 IP 层的长度，然后设置网络头部，接着设置传输层头部的位置，就是一些初始化的动作。

/*
 *	Set up data on packet
 */

ip_copy_metadata(skb2, skb);
skb_reserve(skb2, ll_rs);
skb_put(skb2, len + hlen);
skb_reset_network_header(skb2);
skb2->transport_header = skb2->network_header + hlen;

设置对应 sk 为 owner

/*
 *	Charge the memory for the fragment to any owner
 *	it might possess
 */

if (skb->sk)
	skb_set_owner_w(skb2, skb->sk);

拷贝网络层的头部

/*
 *	Copy the packet header into the new buffer.
 */

skb_copy_from_linear_data(skb, skb_network_header(skb2), hlen);

然后拷贝真正的负载，这里没有直接用 memcpy 的原因是，对应的空间不一定是连续的，它可能含有 frag_list，甚至是之前检查没有通过的快速路径到达了这里。

/*
 *	Copy a block of the IP datagram.
 */
if (skb_copy_bits(skb, ptr, skb_transport_header(skb2), len))
	BUG();
left -= len;

设置 IP 头的偏移和分片标志。

/*
 *	Fill in the new header fields.
 */
iph = ip_hdr(skb2);
iph->frag_off = htons((offset >> 3));

if (IPCB(skb)->flags & IPSKB_FRAG_PMTU)
	iph->frag_off |= htons(IP_DF);

如果是第一个分片就尝试更新 IP options。

/* ANK: dirty, but effective trick. Upgrade options only if
 * the segment to be fragmented was THE FIRST (otherwise,
 * options are already fixed) and make it ONCE
 * on the initial skb, so that all the following fragments
 * will inherit fixed options.
 */
if (offset == 0)
	ip_options_fragment(skb);

最后修改位移，更新标记位，计算 checksum，然后送到 output。

/*
 *	Added AC : If we are fragmenting a fragment that's not the
 *		   last fragment then keep MF on each bit
 */
if (left > 0 || not_last_frag)
	iph->frag_off |= htons(IP_MF);
ptr += len;
offset += len;

/*
 *	Put this fragment into the sending queue.
 */
iph->tot_len = htons(len + hlen);

ip_send_check(iph);

err = output(net, sk, skb2);
if (err)
	goto fail;

IP_INC_STATS(net, IPSTATS_MIB_FRAGCREATES);

这个就是慢速路径的分片过程，快速路径的分片过程其实更简单，因为比较麻烦的事情已经在 ip_append_data 里面处理过了，在我上一篇文章里面有介绍这个过程，就是在上层调用 ip_append_data 的时候，会在主动的进行分段式的缓存，而不使用连续空间，每个分段式的换粗也会不超过分片的大小，这样每个缓存就可以直接用来做分片了。

现在再回头看快速路径，快速路径主要检查有没有 frag_list 也就是之前分配好的 buffer 列表。获取第一个 buffer （存在 frags 里面，不是 frag_list）的长度，如果比 mtu 大，或者不是 8 的倍数，或者已经是分段了，或者是一段 shared skb_buff （因为快速路径不会拷贝内存，慢速路径会会分配新的内存，不影响之前有人引用）都不行，要进入慢速路径。

if (skb_has_frag_list(skb)) {
	struct sk_buff *frag, *frag2;
	unsigned int first_len = skb_pagelen(skb);

	if (first_len - hlen > mtu ||
	    ((first_len - hlen) & 7) ||
	    ip_is_fragment(iph) ||
	    skb_cloned(skb))
		goto slow_path;

首先保证每个 frag_list 里面 frag 不超过 mtu，然后不是最后一段需要是 8 的倍数，有足够的头部空间用来给新的 IP 分片用，然后 frag 的 buffer 也不能是 shared，最后绑定 sk 关系，减掉 skb 的 truesize。

skb_walk_frags(skb, frag) {
	/* Correct geometry. */
	if (frag->len > mtu ||
	    ((frag->len & 7) && frag->next) ||
	    skb_headroom(frag) < hlen)
		goto slow_path_clean;

	/* Partially cloned skb? */
	if (skb_shared(frag))
		goto slow_path_clean;

	BUG_ON(frag->sk);
	if (skb->sk) {
		frag->sk = skb->sk;
		frag->destructor = sock_wfree;
	}
	skb->truesize -= frag->truesize;
}

到这里就可以真的开始分片了，初始化头部信息，以及要用来分片的 frag。

err = 0;
offset = 0;
frag = skb_shinfo(skb)->frag_list;
skb_frag_list_init(skb);
skb->data_len = first_len - skb_headlen(skb);
skb->len = first_len;
iph->tot_len = htons(first_len);
iph->frag_off = htons(IP_MF);
ip_send_check(iph);

这个循环里面做的事情就更简单了，比起慢速路径来说，就是给每个原本没有头部的 buffer，加上头部变成真正的 fragment。保留空间，设置网络层头部，拷贝头部memcpy(skb_network_header(frag), iph, hlen);，拷贝原信息，如果是第一个分片更新 options，然后更新标记位，然后送到 output。直到 frag_list 被循环完，这就大功告成了。

for (;;) {
	/* Prepare header of the next frame,
	 * before previous one went down. */
	if (frag) {
		frag->ip_summed = CHECKSUM_NONE;
		skb_reset_transport_header(frag);
		__skb_push(frag, hlen);
		skb_reset_network_header(frag);
		memcpy(skb_network_header(frag), iph, hlen);
		iph = ip_hdr(frag);
		iph->tot_len = htons(frag->len);
		ip_copy_metadata(frag, skb);
		if (offset == 0)
			ip_options_fragment(frag);
		offset += skb->len - hlen;
		iph->frag_off = htons(offset>>3);
		if (frag->next)
			iph->frag_off |= htons(IP_MF);
		/* Ready, complete checksum */
		ip_send_check(iph);
	}

	err = output(net, sk, skb);

	if (!err)
		IP_INC_STATS(net, IPSTATS_MIB_FRAGCREATES);
	if (err || !frag)
		break;

	skb = frag;
	frag = skb->next;
	skb->next = NULL;
}

IP 重组

重组一般发生在向上层协议栈传输的时候，不过有的路由器也有可能进行重组，可能要对整个 IP packet 进行校验等，一般情况下，转发不太会对 IP 进行重组。IP 重组讲起来也有些麻烦。

每个正在被重组的 IP packet 都会用一个 ipq 表示，这个 ipq 使用的是 hash table (inet_frags->hash) 的搜索结构，没有 ipq 由源地址，目的地址，协议和 ID 确定，所以存在重复的可能。ip_defrag依赖两个函数一个是ip_find用于寻找 ipq 如果没有找到的话会自动创建一个，其次是用于入队的 ip_frag_queue ，进行重组的工作。sk_buff->cb 用于保存当前的 offset。对于分片的重组也会有超时机制，防止一个 ipq 停留太长的时间。

/* Process an incoming IP datagram fragment. */
int ip_defrag(struct net *net, struct sk_buff *skb, u32 user)
{
	struct net_device *dev = skb->dev ? : skb_dst(skb)->dev;
	int vif = l3mdev_master_ifindex_rcu(dev);
	struct ipq *qp;

	__IP_INC_STATS(net, IPSTATS_MIB_REASMREQDS);
	skb_orphan(skb);

	/* Lookup (or create) queue header */
	qp = ip_find(net, ip_hdr(skb), user, vif);
	if (qp) {
		int ret;

		spin_lock(&qp->q.lock);

		ret = ip_frag_queue(qp, skb);

		spin_unlock(&qp->q.lock);
		ipq_put(qp);
		return ret;
	}

	__IP_INC_STATS(net, IPSTATS_MIB_REASMFAILS);
	kfree_skb(skb);
	return -ENOMEM;
}

ip_find 主要两个功能，根据原信息计算 hash 值，从net->ipv4.frags 的 hash 表当中寻找到对应的 ipq。

hash = ipqhashfn(iph->id, iph->saddr, iph->daddr, iph->protocol);

q = inet_frag_find(&net->ipv4.frags, &ip4_frags, &arg, hash);

然后进入到 ip_frag_queue 当中首先检查，如果出现错误，就把 ipq 标记为可以被之后的垃圾回收清扫。

if (qp->q.flags & INET_FRAG_COMPLETE)
	goto err;

if (!(IPCB(skb)->flags & IPSKB_FRAG_COMPLETE) &&
    unlikely(ip_frag_too_far(qp)) &&
    unlikely(err = ip_frag_reinit(qp))) {
	ipq_kill(qp);
	goto err;
}

获取 offset，flags 和头部。

ecn = ip4_frag_ecn(ip_hdr(skb)->tos);
offset = ntohs(ip_hdr(skb)->frag_off);
flags = offset & ~IP_OFFSET;
offset &= IP_OFFSET;
offset <<= 3;		/* offset is in 8-byte chunks */
ihl = ip_hdrlen(skb);

计算这个追加的 fragment 会拷贝的位置的末尾在哪。

/* Determine the position of this fragment. */
end = offset + skb->len - skb_network_offset(skb) - ihl;
err = -EINVAL;

如果是最后一个 fragment，那么不应该超过 q.len，或者已经有了最后一个了，但是 end 和 q.len 不一致，所以有一些 corruption。如果检查没问题，就更新q.flasg 标记为最后一个和把 end 赋值给q.len。

/* Is this the final fragment? */
if ((flags & IP_MF) == 0) {
	/* If we already have some bits beyond end
	 * or have different end, the segment is corrupted.
	 */
	if (end < qp->q.len ||
	    ((qp->q.flags & INET_FRAG_LAST_IN) && end != qp->q.len))
		goto err;
	qp->q.flags |= INET_FRAG_LAST_IN;
	qp->q.len = end;

如果不是最后一个，长度要与 8 对齐，然后更新 q.len。

} else {
	if (end&7) {
		end &= ~7;
		if (skb->ip_summed != CHECKSUM_UNNECESSARY)
			skb->ip_summed = CHECKSUM_NONE;
	}
	if (end > qp->q.len) {
		/* Some bits beyond end -> corruption. */
		if (qp->q.flags & INET_FRAG_LAST_IN)
			goto err;
		qp->q.len = end;
	}

剩下的就是从链表 q.fragments 当中中根据offset 寻找到要插入的位置，会先看一下表尾，再进行遍历。

if (end == offset)
		goto err;

	err = -ENOMEM;
	if (!pskb_pull(skb, skb_network_offset(skb) + ihl))
		goto err;

	err = pskb_trim_rcsum(skb, end - offset);
	if (err)
		goto err;

	/* Find out which fragments are in front and at the back of us
	 * in the chain of fragments so far.  We must know where to put
	 * this fragment, right?
	 */
	prev = qp->q.fragments_tail;
	if (!prev || FRAG_CB(prev)->offset < offset) {
		next = NULL;
		goto found;
	}
	prev = NULL;
	for (next = qp->q.fragments; next != NULL; next = next->next) {
		if (FRAG_CB(next)->offset >= offset)
			break;	/* bingo! */
		prev = next;
	}

如果和前面的分组有重叠，就把重叠的部分去掉，CHECKSUM_NONE 可以使当前的校验和失效。

if (prev) {
	int i = (FRAG_CB(prev)->offset + prev->len) - offset;

	if (i > 0) {
		offset += i;
		err = -EINVAL;
		if (end <= offset)
			goto err;
		err = -ENOMEM;
		if (!pskb_pull(skb, i))
			goto err;
		if (skb->ip_summed != CHECKSUM_UNNECESSARY)
			skb->ip_summed = CHECKSUM_NONE;
	}
}

然后向后检查有没有重叠，并且把重叠的部分去掉，如果重叠的部分比 next 本身还要大，直接把 next 删掉。

while (next && FRAG_CB(next)->offset < end) {
	int i = end - FRAG_CB(next)->offset; /* overlap is 'i' bytes */

	if (i < next->len) {
		/* Eat head of the next overlapped fragment
		 * and leave the loop. The next ones cannot overlap.
		 */
		if (!pskb_pull(next, i))
			goto err;
		FRAG_CB(next)->offset += i;
		qp->q.meat -= i;
		if (next->ip_summed != CHECKSUM_UNNECESSARY)
			next->ip_summed = CHECKSUM_NONE;
		break;
	} else {
		struct sk_buff *free_it = next;

		/* Old fragment is completely overridden with
		 * new one drop it.
		 */
		next = next->next;

		if (prev)
			prev->next = next;
		else
			qp->q.fragments = next;

		qp->q.meat -= free_it->len;
		sub_frag_mem_limit(qp->q.net, free_it->truesize);
		kfree_skb(free_it);
	}
}

剩下的就是插入链表，并且更新 ipq 的信息了。

skb->next = next;
if (!next)
	qp->q.fragments_tail = skb;
if (prev)
	prev->next = skb;
else
	qp->q.fragments = skb;

dev = skb->dev;
if (dev) {
	qp->iif = dev->ifindex;
	skb->dev = NULL;
}
qp->q.stamp = skb->tstamp;
qp->q.meat += skb->len;
qp->ecn |= ecn;
add_frag_mem_limit(qp->q.net, skb->truesize);
if (offset == 0)
	qp->q.flags |= INET_FRAG_FIRST_IN;

fragsize = skb->len + ihl;

if (fragsize > qp->q.max_size)
	qp->q.max_size = fragsize;

if (ip_hdr(skb)->frag_off & htons(IP_DF) &&
    fragsize > qp->max_df_size)
	qp->max_df_size = fragsize;

然后如果，第一个包和最后一个包都收齐了的话，就尝试进行重组。

if (qp->q.flags == (INET_FRAG_FIRST_IN | INET_FRAG_LAST_IN) &&
    qp->q.meat == qp->q.len) {
	unsigned long orefdst = skb->_skb_refdst;

	skb->_skb_refdst = 0UL;
	err = ip_frag_reasm(qp, prev, dev);
	skb->_skb_refdst = orefdst;
	return err;
}

另外垃圾回收的过程，就是在内存超过阈值的时候，把超时的 ipq 从 hash 表当中剔除。内存阈值通过 ip_frag_mem获取。

int ip_frag_mem(struct net *net)
{
	return sum_frag_mem_limit(&net->ipv4.frags);
}

总结

IP 分片与重组的整体流程大致如此，IP 面临的覆盖的现象，是由于不同的 packet 但是 hash 元素一样导致的。另一方面重叠处理一个是防止出现重叠包攻击导致内存溢出。还有就是具体的校验过程会丢给上层的协议来控制。

协议栈 IP 层主体流程

发表于 2017-07-01 更新于 2025-03-14 Disqus：

IP 层主要的工作是头部验证，头部选项的处理，分片和重组，以及路由，本篇文章主要分析 IP 层的主体流程，路由和分片的具体细节暂时略解。

在 ip_init 注册 ip_rcv 处理函数，然后初始化路由子系统，和对端管理器。两个结构 ip_tstamps 和 ip_idents。ip_rcv 是 IP 的入口，主要是一些参数检查。

/*
 * 	Main IP Receive routine.
 */
int ip_rcv(struct sk_buff *skb, struct net_device *dev, struct packet_type *pt, struct net_device *orig_dev)
{
	const struct iphdr *iph;
	struct net *net;
	u32 len;

	/* When the interface is in promisc. mode, drop all the crap
	 * that it receives, do not try to analyse it.
	 */
	 // 如果 L2 地址不是本机地址 pkt_type 就会被设置成 PACKET_OHTERHOST
	 // 然后进行丢包
	if (skb->pkt_type == PACKET_OTHERHOST)
		goto drop;


	net = dev_net(dev);
	__IP_UPD_PO_STATS(net, IPSTATS_MIB_IN, skb->len);
	// 检查引用计数，如果有人引用就复制一份自己的 skb。
	skb = skb_share_check(skb, GFP_ATOMIC);
	if (!skb) {
		__IP_INC_STATS(net, IPSTATS_MIB_INDISCARDS);
		goto out;
	}
	// 保证有 iphdr 大小，如果没有，则可能尝试从 skb_shinfo(skb)->frags[] 中获取
	if (!pskb_may_pull(skb, sizeof(struct iphdr)))
		goto inhdr_error;

	iph = ip_hdr(skb);

	/*
	 *	RFC1122: 3.2.1.2 MUST silently discard any IP frame that fails the checksum.
	 *
	 *	Is the datagram acceptable?
	 *
	 *	1.	Length at least the size of an ip header
	 *	2.	Version of 4
	 *	3.	Checksums correctly. [Speed optimisation for later, skip loopback checksums]
	 *	4.	Doesn't have a bogus length
	 */
	// ip 头部长度至少 20 个字节
	if (iph->ihl < 5 || iph->version != 4)
		goto inhdr_error;

	BUILD_BUG_ON(IPSTATS_MIB_ECT1PKTS != IPSTATS_MIB_NOECTPKTS + INET_ECN_ECT_1);
	BUILD_BUG_ON(IPSTATS_MIB_ECT0PKTS != IPSTATS_MIB_NOECTPKTS + INET_ECN_ECT_0);
	BUILD_BUG_ON(IPSTATS_MIB_CEPKTS != IPSTATS_MIB_NOECTPKTS + INET_ECN_CE);
	__IP_ADD_STATS(net,
		       IPSTATS_MIB_NOECTPKTS + (iph->tos & INET_ECN_MASK),
		       max_t(unsigned short, 1, skb_shinfo(skb)->gso_segs));
	// 保证完整的头部大小
	if (!pskb_may_pull(skb, iph->ihl*4))
		goto inhdr_error;

	iph = ip_hdr(skb);
	// 做校验和
	if (unlikely(ip_fast_csum((u8 *)iph, iph->ihl)))
		goto csum_error;
	// 保证 skb buffer 的大小比 packet 长度大，不然就丢包
    // 这个原因是 L2 有 padding? (TODO)
    // 并且 packet 长度至少有头部那么大
	len = ntohs(iph->tot_len);
	if (skb->len < len) {
		__IP_INC_STATS(net, IPSTATS_MIB_INTRUNCATEDPKTS);
		goto drop;
	} else if (len < (iph->ihl*4))
		goto inhdr_error;

	/* Our transport medium may have padded the buffer out. Now we know it
	 * is IP we can trim to the true length of the frame.
	 * Note this now means skb->len holds ntohs(iph->tot_len).
	 */
	if (pskb_trim_rcsum(skb, len)) {
		__IP_INC_STATS(net, IPSTATS_MIB_INDISCARDS);
		goto drop;
	}

	skb->transport_header = skb->network_header + iph->ihl*4;

	/* Remove any debris in the socket control block */
	memset(IPCB(skb), 0, sizeof(struct inet_skb_parm));
	IPCB(skb)->iif = skb->skb_iif;

	/* Must drop socket now because of tproxy. */
	skb_orphan(skb);

	return NF_HOOK(NFPROTO_IPV4, NF_INET_PRE_ROUTING,
		       net, NULL, skb, dev, NULL,
		       ip_rcv_finish);

csum_error:
	__IP_INC_STATS(net, IPSTATS_MIB_CSUMERRORS);
inhdr_error:
	__IP_INC_STATS(net, IPSTATS_MIB_INHDRERRORS);
drop:
	kfree_skb(skb);
out:
	return NET_RX_DROP;
}

但实际上，大部分的函数是分成两部分的，真正的行为都在 _finish 后缀当中，前期都是检查。在 netfilter 里面 NF_HOOK 这个宏是有个 okfn 如果通过了 netfilter 的检查就会调用这个函数。对应的就是 ip_rcv_finish 当中，要决定是否继续向上层传递还是要进行找到出口设备确定下一跳，进行转发。如果是从设备就交给主设备的 handler 处理（这个和 VRF 有关，主设备代表这些从设备表示的一个域，用于分配一个专有的 FIB 表等等，类似某种程度的隔离）。

/* if ingress device is enslaved to an L3 master device pass the
 * skb to its handler for processing
 */
skb = l3mdev_ip_rcv(skb);
if (!skb)
	return NET_RX_SUCCESS;

如果设置了 ip_early_demux 并且不是分片的 IP 包，就会提前调用 TCP 的 early_demux 提前解复用这个包。

if (net->ipv4.sysctl_ip_early_demux &&
    !skb_dst(skb) &&
    !skb->sk &&
    !ip_is_fragment(iph)) {
	const struct net_protocol *ipprot;
	int protocol = iph->protocol;

	ipprot = rcu_dereference(inet_protos[protocol]);
	if (ipprot && (edemux = READ_ONCE(ipprot->early_demux))) {
		edemux(skb);
		/* must reload iph, skb->head might have changed */
		iph = ip_hdr(skb);
	}
}

在路由系统中找到 dst 指向的 dst_entry，接下来的处理函数也会存在 dst 当中，为一下三种。

ip_forward() 转发到其他主机
ip_local_deliver() 传入传输层
ip_error() 出现了错误，可能会发送一个 ICMP


/*
 *	Initialise the virtual path cache for the packet. It describes
 *	how the packet travels inside Linux networking.
 */
if (!skb_valid_dst(skb)) {
	int err = ip_route_input_noref(skb, iph->daddr, iph->saddr,
				       iph->tos, dev);
	if (unlikely(err)) {
		if (err == -EXDEV)
			__NET_INC_STATS(net, LINUX_MIB_IPRPFILTER);
		goto drop;
	}
}

如果编译选项带了 CONFIG_IP_ROUTE_CLASSID 那么有流量控制的 classid 的就会进行一些统计。

#ifdef CONFIG_IP_ROUTE_CLASSID
	if (unlikely(skb_dst(skb)->tclassid)) {
		struct ip_rt_acct *st = this_cpu_ptr(ip_rt_acct);
		u32 idx = skb_dst(skb)->tclassid;
		st[idx&0xFF].o_packets++;
		st[idx&0xFF].o_bytes += skb->len;
		st[(idx>>16)&0xFF].i_packets++;
		st[(idx>>16)&0xFF].i_bytes += skb->len;
	}
#endif

首先如果 IP 头部长度大于 5 说明有 options，调用 ip_rcv_options 进行处理，如果失败了就进行丢包。

1 2	if (iph->ihl > 5 && ip_rcv_options(skb)) goto drop;

具体过程在 ip_rcv_options 当中。

if (skb_cow(skb, skb_headroom(skb))) {
	__IP_INC_STATS(dev_net(dev), IPSTATS_MIB_INDISCARDS);
	goto drop;
}

首先把 skb_headroom 等于 skb->data - skb->head，计算了头部的长度，如果这个头部有 clone 就会被复制一份来 declone。

然后根据把 IP option 存进结构化的 inet_skb_parm 当中，其中有个成员是 struct ip_options 。IPCB

代表的是 #define IPCB(skb) ((struct inet_skb_parm*)((skb)->cb))，skb->cb 是一个缓冲区用于协议栈每层的处理函数存放一些临时的私有变量。

iph = ip_hdr(skb);
opt = &(IPCB(skb)->opt);
opt->optlen = iph->ihl*4 - sizeof(struct iphdr);

if (ip_options_compile(dev_net(dev), opt, skb)) {
	__IP_INC_STATS(dev_net(dev), IPSTATS_MIB_INHDRERRORS);
	goto drop;
}

in_device 是和 IP 有关的设备信息，如果没有 source route 选项就直接跳过，不然处理 source route 选项。

if (unlikely(opt->srr)) {
	struct in_device *in_dev = __in_dev_get_rcu(dev);

	if (in_dev) {
		if (!IN_DEV_SOURCE_ROUTE(in_dev)) {
			if (IN_DEV_LOG_MARTIANS(in_dev))
				net_info_ratelimited("source route option %pI4 -> %pI4\n",
						     &iph->saddr,
						     &iph->daddr);
			goto drop;
		}
	}

	if (ip_options_rcv_srr(skb))
		goto drop;
}

source route 是一个多字节选项。此选项中，发送节点会列出后续几跳的 IP 地址（不能超过 IP 报头的最大长度）。如果列表中有某台主机宕机了，则必须重新计算来源地路由，重新发送，而不是使用动态路由。ip_options_rcv_srr 的具体工作就是根据提取出的目的地在本地计算目的地是否可达，如果成功就反回 0，不然就丢弃。

从ip_rcv_options 出来以后根据组播广播进行数据统计。下面的 IN_DEV_ORCONF 不太确定是啥 (TODO)。

rt = skb_rtable(skb);
if (rt->rt_type == RTN_MULTICAST) {
	__IP_UPD_PO_STATS(net, IPSTATS_MIB_INMCAST, skb->len);
} else if (rt->rt_type == RTN_BROADCAST) {
	__IP_UPD_PO_STATS(net, IPSTATS_MIB_INBCAST, skb->len);
} else if (skb->pkt_type == PACKET_BROADCAST ||
	   skb->pkt_type == PACKET_MULTICAST) {
	struct in_device *in_dev = __in_dev_get_rcu(dev);

	/* RFC 1122 3.3.6:
	 *
	 *   When a host sends a datagram to a link-layer broadcast
	 *   address, the IP destination address MUST be a legal IP
	 *   broadcast or IP multicast address.
	 *
	 *   A host SHOULD silently discard a datagram that is received
	 *   via a link-layer broadcast (see Section 2.4) but does not
	 *   specify an IP multicast or broadcast destination address.
	 *
	 * This doesn't explicitly say L2 *broadcast*, but broadcast is
	 * in a way a form of multicast and the most common use case for
	 * this is 802.11 protecting against cross-station spoofing (the
	 * so-called "hole-196" attack) so do it for both.
	 */
	if (in_dev &&
	    IN_DEV_ORCONF(in_dev, DROP_UNICAST_IN_L2_MULTICAST))
		goto drop;
}

return dst_input(skb);

进入 dst_input 就会交给 skb>dst >input 来处理。

转发会分成两部分处理 ip_forward 和 ip_forward_finish IP 转发分成几个步骤

处理 IP options，可能会要求记录本地 IP 地址和时间戳
基于 IP 头，确保这个 pakcet 可以发出去
减 1 TTL，到达 0 就丢弃
根据 MTU 进行分组
发送至出口设备

期间如果出错了，会通过 ICMP 告知。xfrm4_xxx 是 IPsec 相关的函数。

1 2	if (IPCB(skb)->opt.router_alert && ip_call_ra_chain(skb)) return NET_RX_SUCCESS;

首先是检查 IP 选项当中有没有 Router Alert，如果有的话就交给 ip_ra_chain 中对此感兴趣的 raw socket 来处理，并且就次结束。

在这里检查 TTL 是否耗尽

1
2
3

if (ip_hdr(skb)->ttl <= 1)
	goto too_many_hops;

如果是严格的源路由，下一条是网关而不是直接连接的路由就丢包。

rt_uses_gateway 代表两种意思

1 的时候表示网关
0 的时候表示直接路由

1
2
3

if (opt->is_strictroute && rt->rt_uses_gateway)
	goto sr_failed;

如果超出 MTU 进行丢包

IPCB(skb)->flags |= IPSKB_FORWARDED;
mtu = ip_dst_mtu_maybe_forward(&rt->dst, true);
if (ip_exceeds_mtu(skb, mtu)) {
	IP_INC_STATS(net, IPSTATS_MIB_FRAGFAILS);
	icmp_send(skb, ICMP_DEST_UNREACH, ICMP_FRAG_NEEDED,
		  htonl(mtu));
	goto drop;
}

declone 这个 skb 并且确保预留 L2 的空间，然后减少 TTL。

/* We are about to mangle packet. Copy it! */
if (skb_cow(skb, LL_RESERVED_SPACE(rt->dst.dev)+rt->dst.header_len))
	goto drop;
iph = ip_hdr(skb);

/* Decrease ttl after skb cow done */
ip_decrease_ttl(iph);

如果被标记为 IPSKB_DOREDIRECT 发送 redirect ICMP，接着设置优先级。

/*
 *	We now generate an ICMP HOST REDIRECT giving the route
 *	we calculated.
 */
if (IPCB(skb)->flags & IPSKB_DOREDIRECT && !opt->srr &&
    !skb_sec_path(skb))
	ip_rt_send_redirect(skb);

skb->priority = rt_tos2priority(iph->tos);

然后进入 ip_forward_finish

1
2
3

return NF_HOOK(NFPROTO_IPV4, NF_INET_FORWARD,
	       net, NULL, skb, skb->dev, rt->dst.dev,
	       ip_forward_finish);

在 ip_forward_finish 当中会进入 dst_output，ip_forward_options 会处理一些 IP 选项并且重新计算 IP 头的校验和。


if (unlikely(opt->optlen))
	ip_forward_options(skb);

return dst_output(net, sk, skb);

在内部有两种情况，一种是单播，一种是广播，对应的处理函数分别是 ip_output 喝 ip_mc_output 两种处理函数，会进行分组操作，然后在 ip_finish_output 当中进入邻居系统。

1	skb_dst(skb)->output(net, sk, skb);

ip_local_deliver 主要的工作是对分片进行重组。

int ip_local_deliver(struct sk_buff *skb)
{
	/*
	 *	Reassemble IP fragments.
	 */
	struct net *net = dev_net(skb->dev);

	if (ip_is_fragment(ip_hdr(skb))) {
		if (ip_defrag(net, skb, IP_DEFRAG_LOCAL_DELIVER))
			return 0;
	}

	return NF_HOOK(NFPROTO_IPV4, NF_INET_LOCAL_IN,
		       net, NULL, skb, skb->dev, NULL,
		       ip_local_deliver_finish);
}

这是 3 层接收端的一个大体结构，下面看一下 3 层发送端的一些内容。发送的入口

1 2	int ip_queue_xmit(struct sk_buff *skb, int ipfragok)

首先会检查是否已经有了路由信息，这个在 SCTP 的情况下会发生。

rt = skb_rtable(skb);
if (rt)
	goto packet_routed;

检查是否缓存了 route 如果有的话却是路由信息的有效性

1 2	/* Make sure we can route this packet. / rt = (struct rtable )__sk_dst_check(sk, 0);

ip_route_output_ports 确保 source route list 的下一跳和 daddr 一致，并且将路由设置在 skb 里面。

if (!rt) {
	__be32 daddr;

	/* Use correct destination address if we have options. */
	daddr = inet->inet_daddr;
	if (inet_opt && inet_opt->opt.srr)
		daddr = inet_opt->opt.faddr;

	/* If this fails, retransmit mechanism of transport layer will
	 * keep trying until route appears or the connection times
	 * itself out.
	 */
	rt = ip_route_output_ports(net, fl4, sk,
				   daddr, inet->inet_saddr,
				   inet->inet_dport,
				   inet->inet_sport,
				   sk->sk_protocol,
				   RT_CONN_FLAGS(sk),
				   sk->sk_bound_dev_if);
	if (IS_ERR(rt))
		goto no_route;
	sk_setup_caps(sk, &rt->dst);
}
skb_dst_set_noref(skb, &rt->dst);

接下来就需要构建 IP 头了，先调用空出需要的 IP 头部空间，并且重置我的大脑。

1 2	/* OK, we know where to send it, allocate and build IP header. */ skb_push(skb, sizeof(struct iphdr) + (inet_opt ? inet_opt->opt.optlen : 0));

重置 IP 头长度其实就是设置了 network_header 等于 data - head

1	skb_reset_network_header(skb);

转换成 16 位的指针并且把 IP 协议号，IP 头长度，inet->tos TOS 写入。

1 2	iph = ip_hdr(skb); ((__be16 )iph) = htons((4 << 12) \| (5 << 8) \| (inet->tos & 0xff));

初始化 iph

if (ip_dont_fragment(sk, &rt->dst) && !skb->ignore_df)
	iph->frag_off = htons(IP_DF);
else
	iph->frag_off = 0;
iph->ttl      = ip_select_ttl(inet, &rt->dst);
iph->protocol = sk->sk_protocol;
ip_copy_addrs(iph, fl4);

如果有选项长度更新选项长度

/* Transport layer set skb->h.foo itself. */

if (inet_opt && inet_opt->opt.optlen) {
	iph->ihl += inet_opt->opt.optlen >> 2;
	ip_options_build(skb, &inet_opt->opt, inet->inet_daddr, rt, 0);
}

基于是否分片给 packet 分配 ID，然后进入 ip_local_out

ip_select_ident_segs(net, skb, sk,
		     skb_shinfo(skb)->gso_segs ?: 1);

/* TODO : should we use skb->sk here instead of sk ? */
skb->priority = sk->sk_priority;
skb->mark = sk->sk_mark;

res = ip_local_out(net, sk, skb);

当这里先告一段落，再来看一下 ip_append_data 这个函数的作用是缓存进合理的结构以便之后进行分片然后发送。ip_push_pending_frames 就可以触发这个动作。

int ip_append_data(struct sock *sk, struct flowi4 *fl4,
		   int getfrag(void *from, char *to, int offset, int len,
			       int odd, struct sk_buff *skb),
		   void *from, int len, int protolen,
		   struct ipcm_cookie *ipc,
		   struct rtable **rt,
		   unsigned int flags);

第一件事情，检查是否有 MSG_PROBE 的标志，有了这个标志的话，表示请求并不真的是需要向下调用。这个在测试对应 IP 地址的 PMTU 的时候会用到。

1
2
3

if (flags&MSG_PROBE)
	return 0;

如果 sock 相关联的 sk_write_queue 队列为空，说明这个是第一个 IP fragment ，如果不是第一个那么就把 transhdrlen 设置成 0，因为只有第一个 IP fragment 才有头部长度的信息。

if (skb_queue_empty(&sk->sk_write_queue)) {
	err = ip_setup_cork(sk, &inet->cork.base, ipc, rtp);
	if (err)
		return err;
} else {
	transhdrlen = 0;
}

ip_setup_cork 主要是初始化了 net->cork，cork 保存了 IP options 和路由信息。

进入 __append_ip_data 之后可以看到

   skb = skb_peek_tail(queue);

exthdrlen = !skb ? rt->dst.header_len : 0;
mtu = cork->fragsize;
if (cork->tx_flags & SKBTX_ANY_SW_TSTAMP &&
    sk->sk_tsflags & SOF_TIMESTAMPING_OPT_ID)
	tskey = sk->sk_tskey++;

计算 L2 头部长度，fragment 头部长度，最大可以容纳的 fragment 长度，还有不分片的情况下的最大长度。（如果忽略分片就直接默认最大值，64KB 就是 0xFFFF，否则使用 mtu）

hh_len = LL_RESERVED_SPACE(rt->dst.dev);

fragheaderlen = sizeof(struct iphdr) + (opt ? opt->optlen : 0);
maxfraglen = ((mtu - fragheaderlen) & ~7) + fragheaderlen;
maxnonfragsize = ip_sk_ignore_df(sk) ? 0xFFFF : mtu;

首先累加的 buffer 长度不能超过最大的 IP 包长度。

if (cork->length + length > maxnonfragsize - fragheaderlen) {
	ip_local_error(sk, EMSGSIZE, fl4->daddr, inet->inet_dport,
		       mtu - (opt ? opt->optlen : 0));
	return -EMSGSIZE;
}

如果是第一个包，并且不会有新的分片，并且硬件支持 checksum 就可以标志为 CHECKSUM_PARTIAL。

/*
 * transhdrlen > 0 means that this is the first fragment and we wish
 * it won't be fragmented in the future.
 */
if (transhdrlen &&
    length + fragheaderlen <= mtu &&
    rt->dst.dev->features & (NETIF_F_HW_CSUM | NETIF_F_IP_CSUM) &&
    !(flags & MSG_MORE) &&
    !exthdrlen)
	csummode = CHECKSUM_PARTIAL;

下面如果满足下面条件就进入 UDP Fragment Offload 例程。是硬件网卡提供的一种特性，由内核和驱动配合完成相关功能。其目的是由网卡硬件来完成本来需要软件进行的分段（分片）操作用于提升效率和性能。如大家所知，在网络上传输的数据包不能大于 mtu，当用户发送大于 mtu 的数据报文时，通常会在传输层（或者在特殊情况下在 IP 层分片，比如 ip 转发或 ipsec 时）就会按 mtu 大小进行分段，防止发送出去的报文大于 mtu，为提升该操作的性能，新的网卡硬件基本都实现了 UFO 功能，可以使分段（或分片）操作在网卡硬件完成，此时用户态就可以发送长度大于 mtu 的包，而且不必在协议栈中进行分段（或分片）。如果硬件支持，是 UDP 协议，并且是大于 mtu 的可以直接用这个函数。

if ((((length + fragheaderlen) > mtu) || (skb && skb_is_gso(skb))) &&
    (sk->sk_protocol == IPPROTO_UDP) &&
    (rt->dst.dev->features & NETIF_F_UFO) && !dst_xfrm(&rt->dst) &&
    (sk->sk_type == SOCK_DGRAM) && !sk->sk_no_check_tx) {
	err = ip_ufo_append_data(sk, queue, getfrag, from, length,
				 hh_len, fragheaderlen, transhdrlen,
				 maxfraglen, flags);
	if (err)
		goto error;
	return 0;
}

剩下的代码有点啰嗦，总体来说就是把 buff 拆分成可以直接发送的 IP fragment，但是需要先把道理讲清楚，不然看代码有点复杂。

确定剩余可以用来拷贝的空间，不能超过 mtu 和 maxfraglen

/* Check if the remaining data fits into current packet. */
	copy = mtu - skb->len;
	if (copy < length)
		copy = maxfraglen - skb->len;

如果不够，就需要分配一个新的 skb，这里面的几个变量具体解释一下，首先是

fraggap 表示的是 mtu 不是 8 的倍数，在最后那个比 8 的倍数多，又小于 mtu 的部分就是 fraggap 了，所以 datalen 是 length + fraggap，fraggap 这部分会从 prev_skb 尾部移动到新 skb 的头部。fraglen 是带上 frag 头部的长度 fraglen = datalen + fragheaderlen。如果 flag 包含了 MSG_MORE 那么会尽量分配一个 mtu，当然这是在不支持 SG(Scatter/Gather I/O) 的情况下。因为支持 SG 的话，就可以直接分散的分配这些内存，最后进行 skb 的分配。

		if (copy <= 0) {
			char *data;
			unsigned int datalen;
			unsigned int fraglen;
			unsigned int fraggap;
			unsigned int alloclen;
			struct sk_buff *skb_prev;
alloc_new_skb:
			skb_prev = skb;
			if (skb_prev)
				fraggap = skb_prev->len - maxfraglen;
			else
				fraggap = 0;

			/*
			 * If remaining data exceeds the mtu,
			 * we know we need more fragment(s).
			 */
			datalen = length + fraggap;
			if (datalen > mtu - fragheaderlen)
				datalen = maxfraglen - fragheaderlen;
			fraglen = datalen + fragheaderlen;

			if ((flags & MSG_MORE) &&
			    !(rt->dst.dev->features&NETIF_F_SG))
				alloclen = mtu;
			else
				alloclen = fraglen;

			alloclen += exthdrlen;

			/* The last fragment gets additional space at tail.
			 * Note, with MSG_MORE we overallocate on fragments,
			 * because we have no idea what fragment will be
			 * the last.
			 */
			if (datalen == length + fraggap)
				alloclen += rt->dst.trailer_len;

			if (transhdrlen) {
				skb = sock_alloc_send_skb(sk,
						alloclen + hh_len + 15,
						(flags & MSG_DONTWAIT), &err);
			} else {
				skb = NULL;
				if (atomic_read(&sk->sk_wmem_alloc) <=
				    2 * sk->sk_sndbuf)
					skb = sock_wmalloc(sk,
							   alloclen + hh_len + 15, 1,
							   sk->sk_allocation);
				if (unlikely(!skb))
					err = -ENOBUFS;
			}
			if (!skb)
				goto error;

length 表示需要传输的长度，在循环不断进行中这个 length 就会变成 0。

接下来的部分是初始化 csum 和 ip_summed 保留硬件头部长度。

         /*
 *	Fill in the control structures
 */
skb->ip_summed = csummode;
skb->csum = 0;
skb_reserve(skb, hh_len);

设置 tx_flags 和 tskey

/* only the initial fragment is time stamped */
			skb_shinfo(skb)->tx_flags = cork->tx_flags;
			cork->tx_flags = 0;
			skb_shinfo(skb)->tskey = tskey;
			tskey = 0;

保留数据空间，并且把指针移动到头部后面，指向负载的部分。

 *	Find where to start putting bytes.
 */
data = skb_put(skb, fraglen + exthdrlen);
skb_set_network_header(skb, exthdrlen);
skb->transport_header = (skb->network_header +
			 fragheaderlen);
data += fragheaderlen + exthdrlen;

这部分就是把上个 skb 的 fraggap 移到当前这个，并且重新获取 checksum。

if (fraggap) {
	skb->csum = skb_copy_and_csum_bits(
		skb_prev, maxfraglen,
		data + transhdrlen, fraggap, 0);
	skb_prev->csum = csum_sub(skb_prev->csum,
				  skb->csum);
	data += fraggap;
	pskb_trim_unique(skb_prev, maxfraglen);
}

接下来调用 getfrag 拷贝数据，加入到队尾。

getfrag 对应的 4 个 routine 分别是

ICMP icmp_glue_bits
UDP. ip_generic_getfrag
RAW iP ip_generic_getfrag
TCP. Ip_reply_glue_bits

getfrag 的功能就是从 from 拷贝到 to ，因为可能是用户态的数据，所以包含了地址转换的功能并且在比好的时候重新计算校验和。

	copy = datalen - transhdrlen - fraggap;
	if (copy > 0 && getfrag(from, data + transhdrlen, offset, copy, fraggap, skb) < 0) {
		err = -EFAULT;
		kfree_skb(skb);
		goto error;
	}

	offset += copy;
	length -= datalen - fraggap;
	transhdrlen = 0;
	exthdrlen = 0;
	csummode = CHECKSUM_NONE;

	if ((flags & MSG_CONFIRM) && !skb_prev)
		skb_set_dst_pending_confirm(skb, 1);

	/*
	 * Put the packet on the pending queue.
	 */
	__skb_queue_tail(queue, skb);
	continue;
}

剩下的就是 copy 足够，不需要分配新的 skb 的条件。

if (copy > length)
	copy = length;

if (!(rt->dst.dev->features&NETIF_F_SG)) {
	unsigned int off;

	off = skb->len;
	if (getfrag(from, skb_put(skb, copy),
			offset, copy, off, skb) < 0) {
		__skb_trim(skb, off);
		err = -EFAULT;
		goto error;
	}
} else {
	int i = skb_shinfo(skb)->nr_frags;

	err = -ENOMEM;
	if (!sk_page_frag_refill(sk, pfrag))
		goto error;

	if (!skb_can_coalesce(skb, i, pfrag->page,
			      pfrag->offset)) {
		err = -EMSGSIZE;
		if (i == MAX_SKB_FRAGS)
			goto error;

		__skb_fill_page_desc(skb, i, pfrag->page,
				     pfrag->offset, 0);
		skb_shinfo(skb)->nr_frags = ++i;
		get_page(pfrag->page);
	}
	copy = min_t(int, copy, pfrag->size - pfrag->offset);
	if (getfrag(from,
		    page_address(pfrag->page) + pfrag->offset,
		    offset, copy, skb->len, skb) < 0)
		goto error_efault;

	pfrag->offset += copy;
	skb_frag_size_add(&skb_shinfo(skb)->frags[i - 1], copy);
	skb->len += copy;
	skb->data_len += copy;
	skb->truesize += copy;
	atomic_add(copy, &sk->sk_wmem_alloc);
}
offset += copy;
length -= copy;

VFIO ——将 DMA 映射暴露给用户态

发表于 2017-06-05 更新于 2025-03-14 Disqus：

VFIO——将设备暴露到用户态

在开始之前我们先要说一个东西就是 DMA，直接让设备访问内存，可以不通过 CPU 搬运数据。

这是一个比较简单的体系结构图，设备和 CPU 通过存储控制器访问存储器。一个简单的 case 是 CPU 向存储器写数据，然后设备从存储器读数据。这么快来一切都很正常。但是实际上 CPU 是有一层缓存的，例如下面这样的。

CPU 想内存写数据，但是先要清空到不一致的缓存，然后设备再去读数据，不然设备读到的数据和 CPU 实际的数据会不一致（因为缓存里的数据可能和存储器的不一致），而且实际上缓存也不只是一层，所以需要一个中间层来保证从 CPU 的角度和从设备的角度内存都是一致的，所以就有了下面这个结构。

CPU 和设备都会走缓存验证一遍以后，再落到存储器上，这样带上缓存以后大家的一致性都是一样的了。所以从设备的角度，设备也拥有了缓存，实际上这个和 IOMMU 关系不是很大，接下来设备其实也可以和 CPU 一样有一层 MMU，也就是地址到存储器物理地址的转换。注意，这里我用了地址，因为对 CPU 来说是虚拟地址，但是对设备来说是一个总线域的地址。这里要明确区分一下，一个是总线地址，是从设备的角度来看的，一个是 CPU 的虚拟地址，这是从 CPU 角度来看的，两个是不同的东西。将总线域地址转换成存储器物理地址的设备就叫 IOMMU。

如果没有 IOMMU，DMA 也能照常工作，IOMMU 的主要作用就是保护功能，防止使用 DMA 的设备访问任意存储器的物理地址。

IOMMU 在不同架构上名字不太一样，AMD 叫 AMD-Vi，最开始针对的设备只是显卡，Intel 叫 VT-d，arm 叫 SMMU，具体对应的手册也不太一样，但是主要解决的问题是一致的。在 VTd 中，dmar (DMA remapping) 就是那个 IOMMU 设备，通过中断的方式实现类似 page fault 一样的内存分配行为。DMA 传输是由 CPU 发起的：CPU 会告诉 DMA 控制器，帮忙将 xxx 地方的数据搬到 xxx 地方。CPU 发完指令之后，就当甩手掌柜了。IOMMU 有点像 MMU 是一个将设备地址翻译到内存地址的页表体系，也会有对应的页表，这个东西在虚拟化中也非常有用，可以将原本有软件模拟的设备，用直接的硬件替代，而原本的隔离通过 IOMMU 来完成。如下图所示，原本需要通过软件模拟的驱动设备可以通过 IOMMU 以__安全__的方式来直接把硬件设备分配个用户态的 Guest OS。

理论上讲没有 IOMMU 实际上是可以工作的，但是硬件的角度，设备就拥有了整个存储器的全局视图，这是无论如何都非常不合理的事情，不应该让设备拥有访问任意物理内存的能力。

这里要提的另外一个功能就是对中断的隔离，类似于下面的通过在中断请求中添加标识来重定向中断到对应的中断回调上。

VFIO 的作用就是通过 IOMMU 以安全的方式来将设备的访问直接暴露到用户空间，而不用专门完成某个驱动等待合并到上游或者使用之前的对 IOMMU 没有感知的 UIO 的框架。通过 VFIO 向用户态开放 IOMMU 的功能，编写用户态的驱动。

对于 IOMMU 来说，隔离的级别不一定是单个设备，比如一个后面有几个设备的 PCI 桥，从 PCI 桥角度来说，都是来自 PCI 桥的总线事务。所以 IOMMU 有一个 iommu_group的概念，代表一组与其他设备隔离的设备的集合。

IOMMU 根据手册上讲还有一个域的概念，可以简单理解为一段物理地址的抽象。

在 iommu_group 的层级上，VFIO 封装了一层 container class，这个的作用对应于希望能够在不同的iommu_group 之间共享 TLB 和 page tables，这个就是一个集合的概念，跟容器的那个概念没啥关系，一个集合总归要有个名字。通过把 host 的 device 和 driver 解绑，然后绑定到 VFIO 的 driver 上，就会有个/dev/vfio/$GROUP/ 出现，然后这个 $GROUP代表的就是这个 device 的 iommu_group号，如果要使用 VFIO 就要把这个 group 下的所有 device 都解绑才可以。

通过打开/dev/vfio/vfio就能创建一个 VFIO 的 container，然后再打开/dev/vfio/$GROUP用VFIO_GROUP_SET_CONTAINER ioctl 把文件描述传进去，就把 group 加进去了，如果支持多个 group 共享页表等结构，还可以把相应的 group 也加进去。（再强调一遍这个页表是总线地址到存储器物理地址，IOMMU 管理的那个页表）。

下面举个官方的栗子，获取 PCI 设备 0000:06:0d.0 的 group_id （PCI 命名的规则是 domain:bus:slot.func）

1 2	$ readlink /sys/bus/pci/devices/0000:06:0d.0/iommu_group ../../../../kernel/iommu_groups/26

使用之前需要你已经加载了 VFIO 模块

1	modprobe vfio-pci

解绑 PCI 设备，然后创建一个 container id

$ lspci -n -s 0000:06:0d.0
06:0d.0 0401: 1102:0002 (rev 08)
# echo 0000:06:0d.0 > /sys/bus/pci/devices/0000:06:0d.0/driver/unbind
# echo 1102 0002 > /sys/bus/pci/drivers/vfio-pci/new_id

然后寻找其他同属于一个 group 的设备

$ ls -l /sys/bus/pci/devices/0000:06:0d.0/iommu_group/devices
total 0
lrwxrwxrwx. 1 root root 0 Apr 23 16:13 0000:00:1e.0 ->
	../../../../devices/pci0000:00/0000:00:1e.0
lrwxrwxrwx. 1 root root 0 Apr 23 16:13 0000:06:0d.0 ->
	../../../../devices/pci0000:00/0000:00:1e.0/0000:06:0d.0
lrwxrwxrwx. 1 root root 0 Apr 23 16:13 0000:06:0d.1 ->
	../../../../devices/pci0000:00/0000:00:1e.0/0000:06:0d.1

PCI 桥 0000:00:1e.0 后面挂了两个设备，一个是刚才加进去的 0000:06:0d.0，还有一个是 0000:06:0d.1，通过上面的步奏加进去就可以。

最后一步是让用户有权限使用这个 group。

1	# chown user:user /dev/vfio/26

下面就是一个样例，从用户态使用 VFIO，整个的使用方式是通过 ioctl来获取中断相关信息，以及注册中断处理函数，然后也是通过 ioctl来获取region信息，然后调用相应的mmap函数，让 CPU 可以访问内存。

int container, group, device, i;
struct vfio_group_status group_status =
				{ .argsz = sizeof(group_status) };
struct vfio_iommu_type1_info iommu_info = { .argsz = sizeof(iommu_info) };
struct vfio_iommu_type1_dma_map dma_map = { .argsz = sizeof(dma_map) };
struct vfio_device_info device_info = { .argsz = sizeof(device_info) };

/* Create a new container */
container = open("/dev/vfio/vfio", O_RDWR);

if (ioctl(container, VFIO_GET_API_VERSION) != VFIO_API_VERSION)
	/* Unknown API version */

if (!ioctl(container, VFIO_CHECK_EXTENSION, VFIO_TYPE1_IOMMU))
	/* Doesn't support the IOMMU driver we want. */

/* Open the group */
group = open("/dev/vfio/26", O_RDWR);

/* Test the group is viable and available */
ioctl(group, VFIO_GROUP_GET_STATUS, &group_status);

if (!(group_status.flags & VFIO_GROUP_FLAGS_VIABLE))
	/* Group is not viable (ie, not all devices bound for vfio) */

/* Add the group to the container */
ioctl(group, VFIO_GROUP_SET_CONTAINER, &container);

/* Enable the IOMMU model we want */
ioctl(container, VFIO_SET_IOMMU, VFIO_TYPE1_IOMMU);

/* Get addition IOMMU info */
ioctl(container, VFIO_IOMMU_GET_INFO, &iommu_info);

/* Allocate some space and setup a DMA mapping */
dma_map.vaddr = mmap(0, 1024 * 1024, PROT_READ | PROT_WRITE,
		     MAP_PRIVATE | MAP_ANONYMOUS, 0, 0);
dma_map.size = 1024 * 1024;
dma_map.iova = 0; /* 1MB starting at 0x0 from device view */
dma_map.flags = VFIO_DMA_MAP_FLAG_READ | VFIO_DMA_MAP_FLAG_WRITE;

ioctl(container, VFIO_IOMMU_MAP_DMA, &dma_map);

/* Get a file descriptor for the device */
device = ioctl(group, VFIO_GROUP_GET_DEVICE_FD, "0000:06:0d.0");

/* Test and setup the device */
ioctl(device, VFIO_DEVICE_GET_INFO, &device_info);

for (i = 0; i < device_info.num_regions; i++) {
	struct vfio_region_info reg = { .argsz = sizeof(reg) };

	reg.index = i;

	ioctl(device, VFIO_DEVICE_GET_REGION_INFO, &reg);

	/* Setup mappings... read/write offsets, mmaps
	 * For PCI devices, config space is a region */
}

for (i = 0; i < device_info.num_irqs; i++) {
	struct vfio_irq_info irq = { .argsz = sizeof(irq) };

	irq.index = i;

	ioctl(device, VFIO_DEVICE_GET_IRQ_INFO, &irq);

	/* Setup IRQs... eventfds, VFIO_DEVICE_SET_IRQS */
}

/* Gratuitous device reset and go... */
ioctl(device, VFIO_DEVICE_RESET);

在include/linux/vfio.h里面有完整的 API，这里就简单略过。

在理解了一些基本原理和使用方式之后再来看 VFIO 的代码应该叫就比较容易理解了。

首先是作为 PCI 设备的 probe。主要是通过 vfio_iommu_group_get 分配 iommu_group，然后调用vfio_add_group_dev初始化设备回调接口vfio_pci_ops，而remove就是反过来把对应的结构释放掉就可以。然后再看注册的回调函数结构体。

static const struct vfio_device_ops vfio_pci_ops = {
	.name		= "vfio-pci",
	.open		= vfio_pci_open,
	.release	= vfio_pci_release,
	.ioctl		= vfio_pci_ioctl,
	.read		= vfio_pci_read,
	.write		= vfio_pci_write,
	.mmap		= vfio_pci_mmap,
	.request	= vfio_pci_request,
};

这里分析几个关键的函数，他们会通过file_operations vfio_fops被间接的调用。

首先是 mmap，就是在调用vfio_pci_mmap的时候最终调用remap_pfn_range(vma, vma->vm_start, vma->vm_pgoff, req_len, vma->vm_page_prot); 来将物理内存映射到用户态空间，这就是上面的栗子中 mmap 系统调用的入口，而具体要映射的物理内存是通过一系列pci_resource_xxx宏从 PCI bar 空间读出来的配置。

然后是 ioctl 接口，这个接口比较丰富，也简单的看一下。比如 VFIO_DEVICE_SET_IRQS会通过使用用户态传进来的结构体，调用vfio_pci_set_irqs_ioctl注册中断处理函数。而通过vfio_ioctl_set_iommu会设置 container 的 iommu_group 以及对应的 driver。read/write接口都是用于修改 PCI 配置信息的。

简单的来说，VFIO 的主要工作是把设备通过 IOMMU 映射的 DMA 物理内存地址映射到用户态中，让用户态程序可以自行操纵设备的传输，并且可以保证一定程度的安全，另外可以自行注册中断处理函数，从而在用户态实现设备的驱动程序，通过这样的框架，可以在 DPDK 中充分发挥用户态协议栈的威力。

参考文献

PCI 基本概念
dmar 和 iommu
总线基本概念
《PCI Express 体系结构导读》王齐著
mastering the dma and iommu apis
VFIO - “Virtual Function I/O”
Intel® Virtualization Technology for Directed I/O: Spec
Linux 下 PCI 设备驱动程序开发

cgroup 子系统之 net_cls 和 net_prio

发表于 2017-05-19 更新于 2025-03-14 Disqus：

在分析net_cls和net_prio之前先要解释几个东西，一个是网络的 QoS 以及 netfilter。

网络 QoS

IP 服务模型是尽力而为的，这样的模型不能体现某些流量的重要性，所以诞生了QOS技术，Linux 很早就提供了流量控制接口，命令行工具是tc。

协议栈的 QoS 主要由三部分组成。

qdisc 队列规则（queueing discipline)，class 控制策略，filter 根据 filter 划入具体的控制策略（class）。

一般由流程是这样的，当一个 qdisc 被入队的时候，会循环匹配其中的 filter 如果匹配到了的话，就会将 packet 入队到对应的 class 当中，大部分情况下被 class 的“所有者”代表的 qdisc 入队。一些没有匹配的 packet 会进入默认的 class。下面将会介绍几种常用的 queue discpline。

pfifo

默认 qdisc 就是 pfifo，实现在 net/sched/sch_prio.c。

设有三个优先级的队列，优先级由高到低分别是

“interactive”
“best effort”
“bulk”

先消费 0 队列再消费 1 队列，依次类推，一般的packet都是属于 1。

pfifo

用 IP 的 ToS 可以映射到这些队列，对应的关系如图。

tos

他的实现依赖下面这个结构，bands 一般是 3 个，代表三个不同优先级的队列，然后filter_list是他的过滤器列表，最后prio2band就是上图的ToS To Queue 的映射中的Linux priority 到 Band的那部分，queues保存的是三个fifo的qdisc，也就是三个最简单的队列。

struct prio_sched_data {
	int bands;
	struct tcf_proto __rcu *filter_list;
	u8  prio2band[TC_PRIO_MAX+1];
	struct Qdisc *queues[TCQ_PRIO_BANDS];
};

这里虽然和 IP 的 ToS 暂时没有关系，但是最后cgroup的部分会提到怎么联系起来的。

pfifo enqueue 的时候会调用prio_classify根据skb->priority来选择队列进行入队，dequeue 的时候则 round robin 每次依次从优先级高到低取出一个 packet。

HTB (Hierarchical Token Bucket)

HTB 是一个层级令牌桶的 qdisc，而且可以加入 class，HTB 的整体结构如下，HTB 的作者在这里解释了他的设计。

这里简单解释一下，每个 class 有一个 AR（保障速率）CR（最大速率），P（优先级），level（在树中的层次），quantum（量子，一个动态参数），和实际的速率 R，中间层的 class 通过计算子层的速率获得。

Leaf 没有子节点，并且只有 Leaf 有传输功能的 queue，其他只是帮助构造层级关系。

Mode class 的状态

Red ( R > CR ) 也就是超速
Yellow (R <= CR && R > AR) 也就是合理超速
Green 就是没有超过保障速率

下面我们有几个等式

1	Rc = min(CRc, ARc + Bc) [eq1]

Bc 表示从祖先那里借到的速率，下面这段公式表示的意思是，如果我是当前 prio 最小的，从优先级的兄弟节点中加权平均借到父节点的速率，其他的 prio 更大的节点都不能借到速率。

       Qc Rp
Bc = -----------------------------  iff min[Pi over D(p)]>=Pc  [eq2]
     sum[Qi over D(p) where Pi=Pc]

Bc = 0   otherwise      [eq3]

如果没有父节点的话，Bc 就是 0。这样算出来代表的意义是什么呢，就是说先服务优先级更高的节点，并且按照量子的大小在同优先级的节点中分配速率。

然后我们具体看一下 HTB 调度器是如何工作的。

htb_sch_1

htb_sch_2

每个 class 有个 slot 包含不同的 prio 级别，指向 yellow 的子节点，然后每个层级包含一个 slot 也有不同的 prio 级别指向这一层中 green 的节点，上图展示了的是有两个 prio 的 slot，右边的 self slot 有个白色的是用于 yellow red 的 wait list。

假设当前如图 1 的状态，所有节点都是绿的没有 packet 到来，现在有有两个包到达 C 和 D，然后激活它们，并且它们现在都是绿的，所以 self slot 指向他们，然后因为 D 的优先级更高，所以先把 D 出队。所以你可以发现，出队的顺序很简答，就是按照优先级从 self slot 里面把绿色的包按顺序取出来就可以。

feed3

feed4

然后我们看一下更复杂的情况，在图 3 中，我们从 D 中出队一个 packet(htb_dequeue)，然后htb_charge_class会增加 D 的速率，导致 D 变成 yellow，离开 self slot （通过htb_deactivate_prios和htb_remove_class_from_row)，然后添加到 B 的 slot 里面 (htb_activate_prios)，并且递归向上添加htb_add_class_to_row，D 会在一段时间后进入 self slot 的白色等待区，然后 D 又会变回绿色。现在如果选择的话，就会从 C 出队，因为虽然 C 的优先级低但是 C 不需要借别人的速率。

在图 4，假设 C 已经完全消耗了速率达到了最大限速，这个时候 D 就会开始工作然后把 B 消耗完，B 被消耗了以后就会去消耗 A，从这里就可以看到一个借取的过程。

htb_sch_5

htb_sch_6

现在说一个更复杂一点的例子，在图 5 中，A 已经被消耗光了，E 开始工作，然后 C 也能开始工作，变成图 6 的样子。注意即使 D 没有被使用但是他的优先级还会被 class slot 维持，也就是红线。但是 C 和 E 都是同一个优先级，这样的话，就要使用 DRR 算法（也就是在 RR 算法上给每个变量加一个权重，也就是之前的那个量子 quantom)。然后也可以发现一个 class 可以对不同的优先级（红色和蓝色）保持 active。

下面是一个 HTB 的全貌图，抽象的可以理解成一个从父 class 中根据优先级带权重的分享令牌的一个算法。

htb arch

下面三个值对应的就是三种颜色。

/* used internaly to keep status of single class */
enum htb_cmode {
	HTB_CANT_SEND,		/* class can't send and can't borrow */
	HTB_MAY_BORROW,		/* class can't send but may borrow */
	HTB_CAN_SEND		/* class can send */
};

HTB 的实现依赖于

struct htb_sched {
	struct Qdisc_class_hash clhash;
	int			defcls;		/* class where unclassified flows go to */
	int			rate2quantum;	/* quant = rate / rate2quantum */

	/* filters for qdisc itself */
	struct tcf_proto __rcu	*filter_list;

#define HTB_WARN_TOOMANYEVENTS	0x1
	unsigned int		warned;	/* only one warning */
	int			direct_qlen;
	struct work_struct	work;

	/* non shaped skbs; let them go directly thru */
	struct qdisc_skb_head	direct_queue;
	long			direct_pkts;

	struct qdisc_watchdog	watchdog;

	s64			now;	/* cached dequeue time */
	struct list_head	drops[TC_HTB_NUMPRIO];/* active leaves (for drops) */

	/* time of nearest event per level (row) */
	s64			near_ev_cache[TC_HTB_MAXDEPTH];

	int			row_mask[TC_HTB_MAXDEPTH];

	struct htb_level	hlevel[TC_HTB_MAXDEPTH];
};

其中hlevel对应的就是self slot，而clhash 可以通过 classid 找到 hub_class，其中的->un.inner就是对应的class slot，->un.leaf对应的就是叶子结点。

使用

qdisc 参数

parent major:minor 或者 root。

一个 qdisc 是根节点就是 root，否则其他的情况指定 parent。其中 major:minor 是 class 的 handle id，每个 class 都要指定一个 id 用于标识。

handle major: ，这个语法有点奇怪，是可选的，如果 qdisc 下面还要分类（多个 class)，则需要指定这个 hanlde。对于 root，通常是”1:”。

// handle 是一组用户指定的标示符，格式为 major:minor。
// 如果是一条 queueing discipline，minor 需要一直为 0。
# tc qdisc add dev eth0 root handle 1: htb

// parent 指明该新增的 class 添加到那一个父 handle 上去
// classid 指明该 class handle 的唯一 ID，minor 需要是非零值
// ceil 定义 rate 的上界
# tc class add dev eth0 parent 1:1 classid 1:6 htb rate 256kbit ceil 512kbit

// 新建一个带宽为 100kbps 的 root class, 其 classid 为 1:1
# tc class add dev eth0 parent 1: classid 1:1 htb rate 100kbps ceil 100kbps
// 接着建立两个子 class，指定其 parent 为 1:1，ceil 用来限制子类最大的带宽
# tc class add dev eth0 parent 1:1 classid 1:10 htb rate 40kbps ceil 100kbps
# tc class add dev eth0 parent 1:1 classid 1:11 htb rate 60kbps ceil 100kbps
// 随后建立 filter 指定哪些类型的 packet 进入那个 class
# tc filter add dev eth0 protocol ip parent 1:0 prio 1 u32 match ip src 1.2.3.4 match ip dport 80 0xffff flowid 1:10
# tc filter add dev eth0 protocol ip parent 1:0 prio 1 u32 match ip src 1.2.3.4 flow 1:11
// 最后为这些 class 添加 queuing disciplines
# tc qdisc add dev eth0 parent 1:10 handle 20: pfifo limit 5
# tc qdisc add dev eth0 parent 1:11 handle 30: sfq perturb 10

实现

用 rtnetlink 接口实现 API。

在include/net/sch_generic.h下有Qdisc_ops,Qdisc_class_ops,tcf_proto_ops的定义。

同一个文件下还有Qdisc_class_ops的定义。

qdisc_run首先检查设备的运行状态然后调用__qdisc_run，这个函数的主体就是调用 qdisc_restart, 直到超过限制或者需要让出时间 CPU 了，最后清空 qdisc的 RUNNING状态。

在此结构中，enqueue 和 dequeue 两个函数是整个 QoS 调度的入口函数。其中的Qdisc_class_ops用于对此 Qdisc的 filter list 进行操作，添加删除等，通过对 Qdisc 添加 fliter，filter 对 enqueue 到此 Qdisc 的 pkt 进行分类，从而归类到此 Qdisc 的子 class 中，而每个子 class 都有自己的 Qdisc 进行 pkt enqueue 的管理，因此实现一个树形的 filter 结构。

callgraph Qdisc 在从来自上层的dev_queue_xmit主动发送开始起作用，对出口数据包作限制。

dev_queue_xmit ->
	__dev_queue_xmit ->
    	__dev_xmit_skb ->
    		qdisc->enqueue
    		__qdisc_run ->
    		  qdisc_restart ->
    		    qdisc->dequeue ->
    		      sch_direct_xmit

在enqueue的时候会主动唤起dequeue也可能是硬件发送就绪会唤醒发送软中断来dequeue，描述的整个过程大概是图中的这部分。

补充

netfilter

netfilter 在数据包传输中有一些 hook 可以在其中注册回调函数。

netfilter

iptables

主要是四表五链，是 netfilter 的用户态工具。

deep dive

cgroup 子系统 net_cls (Network classifier cgroup)

net_cls 可以给 packet 打上 classid 的标签，用于过滤分类，有了上面的详细解释，这个 classid 的作用也非常明显了，就是用于标记skb所属的 qdisc class 的。

有了这个标签，流量控制器（tc）可以对不同的 cgroup 的 packet 起作用，Netfilter（iptables）也可以基于这个标签有对应的动作。创建一个 net_cls cgroup 对应的是创建一个 net_cls.classid 文件，这个文件初始化为 0。可以写 16 进制的 0xAAAABBBB 到这个文件里面，AAAA 是 major 号，BBBB 是 minor 号。读这个文件返回的是十进制的数字。

例子

mkdir /sys/fs/cgroup/net_cls
mount -t cgroup -onet_cls net_cls /sys/fs/cgroup/net_cls
mkdir /sys/fs/cgroup/net_cls/0
echo 0x100001 >  /sys/fs/cgroup/net_cls/0/net_cls.classid

设置一个 10:1 handle.

1 2	cat /sys/fs/cgroup/net_cls/0/net_cls.classid 1048577

配置 tc:

1 2	tc qdisc add dev eth0 root handle 10: htb tc class add dev eth0 parent 10: classid 10:1 htb rate 40mbit

创建 traffic class 10:1

1	tc filter add dev eth0 parent 10: protocol ip prio 10 handle 1: cgroup

配置 iptables，也可以用于这个 classid。

1	iptables -A OUTPUT -m cgroup ! --cgroup 0x100001 -j DROP

对应的实现在net/core/netclassid_cgroup.c下面。起作用的方式是css_cls_state的classid并且sock_cgroup_set_classid(&sock->sk->sk_cgrp_data,(unsigned long)v)来设置sock的classid。

cgroup net_prio 子系统

网络优先权（net_prio）子系统可以为各个 cgroup 中的应用程序动态配置每个网络接口的流量优先级。

net_prio.prioidx

只读文件。它包含一个特有整数值，kernel 使用该整数值作为这个 cgroup 的内部代表。

net_prio.ifpriomap

包含优先级图谱，这些优先级被分配给源于此群组进程的流量以及通过不同接口离开系统的流量。回顾pfifo里优先级映射，对应的就是这个值。该图用的形式以成对列表表示：

~]# cat /cgroup/net_prio/iscsi/net_prio.ifpriomap
eth0 5
eth1 4
eth2 6

net_prio.ifpriomap 文件的目录可以使用上述格式，通过将字符串回显至文件的方式来修改。例如：

1	~]# echo "eth0 5" > /cgroup/net_prio/iscsi/net_prio.ifpriomap

上述指令将强制设定任何源于 iscsi net_prio cgroup 进程的流量和 eth0 网络接口传出的流量的优先级为 5。父 cgroup 也有可写入的 net_prio.ifpriomap 文件，可以设定系统默认优先级。

对应的实现在net/core/netprio_cgroup.c下面。实现方式是通过扩展dev->priomap的prioid->prio的映射记录这个优先级和 cgroup 的关系。

net_prio 使用每个 cgroup 的 id（cgroupo->id）作为 sequence number，并将这个存储在 sk_cgrp_prioidx 中。sk_cgrp_prioidx 这个是单纯的用于设置网络包的优先级，使用这个之后将会覆盖之前通过 SO_PRIORITY socket 选项或者其他方式设置的值。