Generic+Netlink內核實現分析（二）：通信

前一篇博文中分析了Generic Netlink的消息結構及內核初始化流程，本文中通過一個示例程序來了解Generic Netlink在內核和應用層之間的單播通信流程。
示例程序：demo_genetlink_kern.c（內核模塊）、demo_genetlink_user.c（應用層Demo程序）、demo_genetlink.h
程序主要功能：應用層程序接收用戶的輸入“字符串”和“數據”向內核發送，內核接收後回發應用層，應用層通過終端打印輸出。
代碼路徑：https://github.com/luckyapple1028/demo-genetlink

一、創建內核Demo Genetlink

1、定義Demo Genetlink

enum {
	DEMO_CMD_UNSPEC = 0,	/* Reserved */
	DEMO_CMD_ECHO,			/* user->kernel request/get-response */
	DEMO_CMD_REPLY,			/* kernel->user event */
	__DEMO_CMD_MAX,
};
#define DEMO_CMD_MAX (__DEMO_CMD_MAX - 1)

enum {
	DEMO_CMD_ATTR_UNSPEC = 0,
	DEMO_CMD_ATTR_MESG,		/* demo message  */
	DEMO_CMD_ATTR_DATA,		/* demo data */
	__DEMO_CMD_ATTR_MAX,
};
#define DEMO_CMD_ATTR_MAX (__DEMO_CMD_ATTR_MAX - 1)

定義兩種類型的Genetlink cmd指令，其中DEMO_CMD_ECHO用於應用層下發數據，DEMO_CMD_REPLY用於內核向應用層回發數據；同時定義兩種類型的attr屬性參數，其中DEMO_CMD_ATTR_MESG表示字符串，DEMO_CMD_ATTR_DATA表示數據。

static struct genl_family demo_family = {
	.id			= GENL_ID_GENERATE,
	.name		= DEMO_GENL_NAME,
	.version	= DEMO_GENL_VERSION,
	.maxattr	= DEMO_CMD_ATTR_MAX,
};

定義demo_family，其中ID號為GENL_ID_GENERATE，表示由內核統一分配，maxattr為DEMO_CMD_ATTR_MAX，其前文中定義的最大attr屬性數，內核將為其分配緩存空間。

static const struct genl_ops demo_ops[] = {
	{
		.cmd		= DEMO_CMD_ECHO,
		.doit		= demo_echo_cmd,
		.policy		= demo_cmd_policy,
		.flags		= GENL_ADMIN_PERM,
	},
};

定義操作函數集operations：demo_ops，這裡只為DEMO_CMD_ECHO類型的cmd創建消息處理函數接口（因為DEMO_CMD_REPLY類型的cmd用於內核消息，應用層不使用），指定doit消息處理回調函數為demo_echo_cmd，同時指定有效組策略為demo_cmd_policy。

static const struct nla_policy demo_cmd_policy[DEMO_CMD_ATTR_MAX+1] = {
	[DEMO_CMD_ATTR_MESG]	= { .type = NLA_STRING },
	[DEMO_CMD_ATTR_DATA]	= { .type = NLA_S32 },	
};

這裡限定DEMO_CMD_ATTR_MESG的屬性類型為NLA_STRING（字符串類型），限定DEMO_CMD_ATTR_DATA的屬性類型為NLA_S32（有符號32位數）。

2、內核注冊Demo Genetlink

static int __init demo_genetlink_init(void) 
{  
	int ret;  
	pr_info("demo generic netlink module %d init...\n", DEMO_GENL_VERSION);  
  
	ret = genl_register_family_with_ops(&demo_family, demo_ops);  
	if (ret != 0) {  
		pr_info("failed to init demo generic netlink example module\n");  
		return ret;
	}  

	pr_info("demo generic netlink module init success\n");  

	return 0; 
}

在模塊的初始化函數中，調用genl_register_family_with_ops()同時注冊demo_family及demo_ops，該函數同前面創建CTRL類型的family簇類似，最終都是調用_genl_register_family_with_ops_grps函數完成創建。這個函數已經大致分析過了，此處的注冊流程基本一致，主要區別在於最後的send all events，它向所有的應用層加入CTRL控制器簇組播組的Generic
Netlink套接字多播發送CTRL_CMD_NEWFAMILY消息，通知應用層有新的family注冊了，這樣應用層就可以捕獲這一消息。詳細分析一下：

static int genl_ctrl_event(int event, struct genl_family *family,
			   const struct genl_multicast_group *grp,
			   int grp_id)
{
	struct sk_buff *msg;

	/* genl is still initialising */
	if (!init_net.genl_sock)
		return 0;

	switch (event) {
	case CTRL_CMD_NEWFAMILY:
	case CTRL_CMD_DELFAMILY:
		WARN_ON(grp);
		msg = ctrl_build_family_msg(family, 0, 0, event);
		break;
	case CTRL_CMD_NEWMCAST_GRP:
	case CTRL_CMD_DELMCAST_GRP:
		BUG_ON(!grp);
		msg = ctrl_build_mcgrp_msg(family, grp, grp_id, 0, 0, event);
		break;
	default:
		return -EINVAL;
	}

函數首先判斷是否已經注冊了控制器CTRL簇，這裡顯然已經注冊過了，然後主要的工作在ctrl_build_family_msg()中：

static struct sk_buff *ctrl_build_family_msg(struct genl_family *family,
					     u32 portid, int seq, u8 cmd)
{
	struct sk_buff *skb;
	int err;

	skb = nlmsg_new(NLMSG_DEFAULT_SIZE, GFP_KERNEL);
	if (skb == NULL)
		return ERR_PTR(-ENOBUFS);

	err = ctrl_fill_info(family, portid, seq, 0, skb, cmd);
	if (err < 0) {
		nlmsg_free(skb);
		return ERR_PTR(err);
	}

	return skb;
}

首先調用nlmsg_new()函數創建netlink類型的skb，第一個入參是消息的長度，第二個參數為內存空間分配類型，這裡分配的數據空間(包括netlink消息頭)一共為一個page。進入nlmsg_new內部：

static inline struct sk_buff *nlmsg_new(size_t payload, gfp_t flags)
{
	return alloc_skb(nlmsg_total_size(payload), flags);
}

static inline int nlmsg_total_size(int payload)
{
	return NLMSG_ALIGN(nlmsg_msg_size(payload));
}

static inline int nlmsg_msg_size(int payload)
{
	return NLMSG_HDRLEN + payload;
}

可以看到總共預留的空間為NLMSG_ALIGN（NLMSG_HDRLEN+NLMSG_DEFAULT_SIZE），這裡實際可能用不了這麼多的空間，接下來調用ctrl_fill_info()填充消息內容：

static int ctrl_fill_info(struct genl_family *family, u32 portid, u32 seq,
			  u32 flags, struct sk_buff *skb, u8 cmd)
{
	void *hdr;

	hdr = genlmsg_put(skb, portid, seq, &genl_ctrl, flags, cmd);
	if (hdr == NULL)
		return -1;

這個函數比較長，這裡使用插圖的形式來觀察消息的封裝流程（圖中未顯示空白Pad區）：

圖1 消息的封裝流程（a）（b）（c）
首先確定各個入參的內容：family為新注冊的demo_family；portid和seq為0，表示消息的發送端為內核，發送消息序號為0；最後的cmd為CTRL_CMD_NEWFAMILY。函數首先調用genlmsg_put()函數初始化netlink消息頭和genetlink消息頭；

void *genlmsg_put(struct sk_buff *skb, u32 portid, u32 seq,
				struct genl_family *family, int flags, u8 cmd)
{
	struct nlmsghdr *nlh;
	struct genlmsghdr *hdr;

	nlh = nlmsg_put(skb, portid, seq, family->id, GENL_HDRLEN +
			family->hdrsize, flags);
	if (nlh == NULL)
		return NULL;

其中nlmsg_put()函數向skb緩沖區中獲取消息頭空間並且初始化netlink消息頭，入參中的第5個參數為genetlink消息頭和用戶私有消息頭（這裡並未使用）的總空間，實際調用的函數為__nlmsg_put()：

struct nlmsghdr *
__nlmsg_put(struct sk_buff *skb, u32 portid, u32 seq, int type, int len, int flags)
{
	struct nlmsghdr *nlh;
	int size = nlmsg_msg_size(len);

	nlh = (struct nlmsghdr *)skb_put(skb, NLMSG_ALIGN(size));
	nlh->nlmsg_type = type;
	nlh->nlmsg_len = size;
	nlh->nlmsg_flags = flags;
	nlh->nlmsg_pid = portid;
	nlh->nlmsg_seq = seq;
	if (!__builtin_constant_p(size) || NLMSG_ALIGN(size) - size != 0)
		memset(nlmsg_data(nlh) + len, 0, NLMSG_ALIGN(size) - size);
	return nlh;
}

首先這裡的分配的空間大小為size = 傳入的len長度 + netlink消息頭的長度，然後初始化netlink消息頭的各個字段：
nlh->nlmsg_type ：內核genl_ctrl family簇的ID 號GENL_ID_CTRL；
nlh->nlmsg_len ：消息長度，即genetlink頭+用戶私有頭+netlink頭的長度總和；
nlh->nlmsg_flags：0；
nlh->nlmsg_pid：發送端的ID號為0，表示又內核發送；
nlh->nlmsg_seq：0；
初始化完成後將內存對齊用的空白區刷為0；然後回到genlmsg_put()函數中繼續分析：

hdr = nlmsg_data(nlh);
	hdr->cmd = cmd;
	hdr->version = family->version;
	hdr->reserved = 0;

	return (char *) hdr + GENL_HDRLEN;
}

這裡通過宏nlmsg_data獲取genetlink消息頭的地址，然後開始填充該消息頭的各個字段：
hdr->cmd：消息的cmd命令，CTRL_CMD_NEWFAMILY；
hdr->version：genl_ctrl family簇的version；
hdr->reserved：0；
填充完畢後返回消息用戶私有頭（若有）或實際載荷的首地址，此時的消息skb中的消息填充如圖1-a所示。然後再回到ctrl_fill_info()函數中，接下來就要開始填充實際的數據了：

if (nla_put_string(skb, CTRL_ATTR_FAMILY_NAME, family->name) ||
	    nla_put_u16(skb, CTRL_ATTR_FAMILY_ID, family->id) ||
	    nla_put_u32(skb, CTRL_ATTR_VERSION, family->version) ||
	    nla_put_u32(skb, CTRL_ATTR_HDRSIZE, family->hdrsize) ||
	    nla_put_u32(skb, CTRL_ATTR_MAXATTR, family->maxattr))
		goto nla_put_failure;

這裡將新注冊的family結構中的幾個字段都填充到了消息中，包括name、id號、版本號、私有頭長度以及maxattr（注意屬性需要一一對應），調用的函數nla_put_string、nla_put_u16和nla_put_u32都是nla_put()的封裝，而nla_put實際調用的是__nla_put()：

int nla_put(struct sk_buff *skb, int attrtype, int attrlen, const void *data)
{
	if (unlikely(skb_tailroom(skb) < nla_total_size(attrlen)))
		return -EMSGSIZE;

	__nla_put(skb, attrtype, attrlen, data);
	return 0;
}

void __nla_put(struct sk_buff *skb, int attrtype, int attrlen,
			     const void *data)
{
	struct nlattr *nla;

	nla = __nla_reserve(skb, attrtype, attrlen);
	memcpy(nla_data(nla), data, attrlen);
}

__nla_put()的作用是向skb中添加一個netlink attr屬性，入參分別為skb地址、要添加的attr屬性類型、屬性長度和屬性實際數據。首先調用了__nla_reserve在skb中預留出attr屬性的內存空間：

struct nlattr *__nla_reserve(struct sk_buff *skb, int attrtype, int attrlen)
{
	struct nlattr *nla;

	nla = (struct nlattr *) skb_put(skb, nla_total_size(attrlen));
	nla->nla_type = attrtype;
	nla->nla_len = nla_attr_size(attrlen);

	memset((unsigned char *) nla + nla->nla_len, 0, nla_padlen(attrlen));

	return nla;
}

這裡首先預留空間長度為nla_total_size(attrlen)，即attrlen+NLA_HDRLEN（屬性頭長度）+對齊用內存空白；然後初始化屬性頭的兩個字段：
nla->nla_type：attr屬性，即前文中的CTRL_ATTR_FAMILY_NAME等；
nla->nla_len：attr屬性長度（attrlen+NLA_HDRLEN）；
然後再將attr屬性中的實際數據拷貝到預留測空間中，如此一個attr屬性就添加完成了，此時的消息skb中的消息填充如圖1-b所示。再回到ctrl_fill_info()函數中：

if (family->n_ops) {
		struct nlattr *nla_ops;
		int i;

		nla_ops = nla_nest_start(skb, CTRL_ATTR_OPS);
		if (nla_ops == NULL)
			goto nla_put_failure;

		for (i = 0; i < family->n_ops; i++) {
			struct nlattr *nest;
			const struct genl_ops *ops = &family->ops[i];
			u32 op_flags = ops->flags;

			if (ops->dumpit)
				op_flags |= GENL_CMD_CAP_DUMP;
			if (ops->doit)
				op_flags |= GENL_CMD_CAP_DO;
			if (ops->policy)
				op_flags |= GENL_CMD_CAP_HASPOL;

			nest = nla_nest_start(skb, i + 1);
			if (nest == NULL)
				goto nla_put_failure;

			if (nla_put_u32(skb, CTRL_ATTR_OP_ID, ops->cmd) ||
			    nla_put_u32(skb, CTRL_ATTR_OP_FLAGS, op_flags))
				goto nla_put_failure;

			nla_nest_end(skb, nest);
		}

		nla_nest_end(skb, nla_ops);
	}

然後如果新注冊的family簇也同時注冊了操作接口operations，這裡會追加上對應的attr屬性參數；但同前面不同的是，這裡追加的attr參數是“打包”在一起的，使用的屬性為CTRL_ATTR_OPS。由於netlink的attr屬性是支持多級嵌套的，所以這裡的“打包”指的就是新建一級嵌套，首先使用nla_nest_start()函數來創建新的一級嵌套：

static inline struct nlattr *nla_nest_start(struct sk_buff *skb, int attrtype)
{
	struct nlattr *start = (struct nlattr *)skb_tail_pointer(skb);

	if (nla_put(skb, attrtype, 0, NULL) < 0)
		return NULL;

	return start;
}

可以看到這裡調用的依然是nla_put()函數，不過這裡的入參中指定的attr長度為0，然後數據為NULL，那這裡其實就是向skb中添加了一段attr屬性頭，然後指定它的屬性nla_type為CTRL_ATTR_OPS，屬性nla_len為0，注意函數返回的是添加嵌套attr頭之前的消息有效數據末尾地址。
然後回到ctrl_fill_info()函數中繼續往下是一個for循環，在每個循環中向剛才新創建的一級嵌套attr屬性中添加屬性。它首先會根據operations中實現的回調函數封裝flag，然後依舊是調用nla_nest_start()函數再次創建新的一級嵌套，不過這次的attrtype為函數的序列，隨後在消息中追加上回調函數處理的cmd以及flag，最後調用nla_nest_end()函數結束這一層attr嵌套：

static inline int nla_nest_end(struct sk_buff *skb, struct nlattr *start)
{
	start->nla_len = skb_tail_pointer(skb) - (unsigned char *)start;
	return skb->len;
}

這個函數其實只做了一件事，那就是更新這個嵌套的attr屬性頭的nla_len字段為本嵌套屬性的實際長度，實現的方式為當前的消息末尾地址減去創建該級嵌套之前的消息末尾地址（這就是nla_nest_start()函數要返回start地址的原因了）。回到ctrl_fill_info()函數中，在for循環結束以後，依舊調用nla_nest_end來結束CTRL_ATTR_OPS的那一層attr嵌套，此時的消息skb中的消息填充如圖1-c所示。
ctrl_fill_info()函數接下來會再判斷family->n_mcgrps字段，若存在組播組，會同operations一樣增加一級和operations平級的attr嵌套然後添加CTRL_ATTR_MCAST_GROUPS屬性，這裡就不詳細分析了。在函數的最後調用nla_nest_end()完成本次消息封裝：

static inline void genlmsg_end(struct sk_buff *skb, void *hdr)
{
	nlmsg_end(skb, hdr - GENL_HDRLEN - NLMSG_HDRLEN);
}

該函數間接調用nlmsg_end()函數，注意第二個入參為消息attr載荷的首地址減去2個頭的長度，即netlink消息頭的首地址。

static inline void nlmsg_end(struct sk_buff *skb, struct nlmsghdr *nlh)
{
	nlh->nlmsg_len = skb_tail_pointer(skb) - (unsigned char *)nlh;
}

這裡填充nlh->nlmsg_len為整個消息的長度(包括attr載荷部分和所有的消息頭部分)。到這裡，向CTRL控制器簇發送的消息就已經封裝完成了，再回到最上層的genl_ctrl_event()中：

if (IS_ERR(msg))
		return PTR_ERR(msg);

	if (!family->netnsok) {
		genlmsg_multicast_netns(&genl_ctrl, &init_net, msg, 0,
					0, GFP_KERNEL);
	} else {
		rcu_read_lock();
		genlmsg_multicast_allns(&genl_ctrl, msg, 0,
					0, GFP_ATOMIC);
		rcu_read_unlock();
	}

這裡根據是否支持net命名空間來選擇發送的流程，genlmsg_multicast_allns函數從命名中就可以看出會像所有命名空間的控制器簇發送消息，而genlmsg_multicast_netns則指定了向init_net發送，不論哪一種情況，最後都是調用nlmsg_multicast()函數。不過這裡有一點需要注意的就是這裡的第三個入參portid為0，這是為了防止向發送端發送報文，這也就表明內核控制器簇套接字是不會接受該廣播報文的（內核也不應該接收，否則會panic，可參見netlink_data_ready()函數的實現）。
至此Demo Genetlink的內核創建流程就全部結束了，此時應用層可以通過Ctrl簇獲取它的ID號並向他發送消息了。下面來分析應用層是如何初始化genetlink套接字的。

二、應用層初始化Genetlink套接字

int main(int argc, char* argv[]) 
{
	......

	/* 初始化socket */	
	nl_fd = demo_create_nl_socket(NETLINK_GENERIC);
	if (nl_fd < 0) {
		fprintf(stderr, "failed to create netlink socket\n");
		return 0;		
	}

	......
}

static int demo_create_nl_socket(int protocol)
{
	int fd;
	struct sockaddr_nl local;

	/* 創建socket */
	fd = socket(AF_NETLINK, SOCK_RAW, protocol);
	if (fd < 0)
		return -1;

	memset(&local, 0, sizeof(local));
	local.nl_family = AF_NETLINK;
	local.nl_pid = getpid();

	/* 使用本進程的pid進行綁定 */
	if (bind(fd, (struct sockaddr *) &local, sizeof(local)) < 0)
		goto error;

	return fd;
	
error:
	close(fd);
	return -1;
}

應用層依然通過socket系統調用創建AF_NETLINK地址簇的SOCK_RAW套接字，指定協議類型為NETLINK_GENERIC，創建的流程同前博文《Netlink 內核實現分析（一）：創建》中分析的NETLINK_ROUTE類似，這裡不再贅述。
接下來初始化sockaddr_nl地址結構並進行綁定操作，為了簡單起見，這裡使用進程ID進行綁定（該值全局唯一），在實際的程序中可自行安排。綁定的流程前博文也已經分析過了，會調用到netlink回調函數netlink_bind()，該函數會將綁定的ID號添加到全局nl_table中。這裡有一點需要說明的就是在前一篇博文中已經看到內核genetlink套接字已經指定了bind回調函數為genl_bind，這裡如果指定了多播組地址nl_groups，會調用到該回調函數進行多播組的綁定操作。簡單看一下：

if (nlk->netlink_bind && groups) {
		int group;

		for (group = 0; group < nlk->ngroups; group++) {
			if (!test_bit(group, &groups))
				continue;
			err = nlk->netlink_bind(net, group + 1);
			if (!err)
				continue;
			netlink_undo_bind(group, groups, sk);
			return err;
		}
	}

這裡在genetlink支持的最大組播數中進行輪詢，檢測用戶需要綁定的多播組並將其轉換為位序號，然後調用netlink_bind回調函數，這裡該函數就是genl_bind()：

static int genl_bind(struct net *net, int group)
{
	int i, err = -ENOENT;

	down_read(&cb_lock);
	for (i = 0; i < GENL_FAM_TAB_SIZE; i++) {
		struct genl_family *f;

		list_for_each_entry(f, genl_family_chain(i), family_list) {
			if (group >= f->mcgrp_offset &&
			    group < f->mcgrp_offset + f->n_mcgrps) {
				int fam_grp = group - f->mcgrp_offset;

				if (!f->netnsok && net != &init_net)
					err = -ENOENT;
				else if (f->mcast_bind)
					err = f->mcast_bind(net, fam_grp);
				else
					err = 0;
				break;
			}
		}
	}
	up_read(&cb_lock);

	return err;
}

由於不同family類型的genetlink都共用同一個組播地址空間，所以這裡根據用戶輸入的組播號來查找對應的family，然後會調用該family對應的mcast_bind()回調函數，它需要根據family的需求自行實現，可用於做進一步的特殊需求處理，不實現亦可（目前內核中注冊的genl_family均未使用到該接口）。
至此，應用層genetlink套接字初始化完成，下面來分析它是如何發送消息到前文中注冊的內核demo genelink套接字的。

三、用戶空間和內核空間通信

用戶空間想要發送消息到內核的demo genelink套接字，它首先得知道內核分配的demo family的family id號，因為genelink子系統是根據該id號來區分不同family簇的genelink套接字和分發消息的。此時前文中的ctrl就用於該目的，它可以將family name轉換為對應的family id，用戶空間也通過family name向ctrl簇查詢對應的family id。在應用層序獲取了family id後它就可以像內核發送消息，該消息分別包含了字符串和數據，同時內核也在接受後進行回發操作。
另外，在一般的程序中，如果應用層無需向內核發送消息，僅僅需要接收內核發送的消息時，它並不需要通過Ctrl簇獲取family id了，僅需要接收內核的genetlink消息並做好cmd和attr類型判斷並做出相應的處理即可。

1、用戶查詢Demo Family ID

圖2 查詢Family ID的調用流程

int main(int argc, char* argv[]) 
{
	......
	
	/* 獲取family id */
	nl_family_id = demo_get_family_id(nl_fd);
	if (!nl_family_id) {
		fprintf(stderr, "Error getting family id, errno %d\n", errno);
		goto out;
	}
	PRINTF("family id %d\n", nl_family_id);

	......
}

static int demo_get_family_id(int sd)
{
	struct msgtemplate ans;
	
	char name[100];
	int id = 0, ret;
	struct nlattr *na;
	int rep_len;

	/* 根據gen family name查詢family id */
	strcpy(name, DEMO_GENL_NAME);
	ret = demo_send_cmd(sd, GENL_ID_CTRL, getpid(), CTRL_CMD_GETFAMILY,
			CTRL_ATTR_FAMILY_NAME, (void *)name, strlen(DEMO_GENL_NAME)+1);
	if (ret < 0)
		return 0;	

	/* 接收內核消息 */
	rep_len = recv(sd, &ans, sizeof(ans), 0);
	if (ans.n.nlmsg_type == NLMSG_ERROR || (rep_len < 0) || !NLMSG_OK((&ans.n), rep_len))
		return 0;

	/* 解析family id */
	na = (struct nlattr *) GENLMSG_DATA(&ans);
	na = (struct nlattr *) ((char *) na + NLA_ALIGN(na->nla_len));
	if (na->nla_type == CTRL_ATTR_FAMILY_ID) {
		id = *(__u16 *) NLA_DATA(na);
	}
	
	return id;
}

該demo_get_family_id()函數比較簡單，僅僅是封裝查詢消息並向內核的ctrl簇發送，然後接收內核的回發結果然後解析出其中的family id，具體的消息發送函數由demo_send_cmd()封裝函數來完成，其中入參分別是socket fd、ctrl family id、消息發送端netlink綁定ID號、消息cmd類型、消息attr屬性、消息正文內容、消息正文長度。

static int demo_send_cmd(int sd, __u16 nlmsg_type, __u32 nlmsg_pid,
	     __u8 genl_cmd, __u16 nla_type,
	     void *nla_data, int nla_len)
{
	struct nlattr *na;
	struct sockaddr_nl nladdr;
	int r, buflen;
	char *buf;

	struct msgtemplate msg;

	/* 填充msg (本函數發送的msg只填充一個attr) */
	msg.n.nlmsg_len = NLMSG_LENGTH(GENL_HDRLEN);
	msg.n.nlmsg_type = nlmsg_type;
	msg.n.nlmsg_flags = NLM_F_REQUEST;
	msg.n.nlmsg_seq = 0;
	msg.n.nlmsg_pid = nlmsg_pid;
	msg.g.cmd = genl_cmd;
	msg.g.version = DEMO_GENL_VERSION;
	na = (struct nlattr *) GENLMSG_DATA(&msg);
	na->nla_type = nla_type;
	na->nla_len = nla_len + 1 + NLA_HDRLEN;
	memcpy(NLA_DATA(na), nla_data, nla_len);
	msg.n.nlmsg_len += NLMSG_ALIGN(na->nla_len);

	buf = (char *) &msg;
	buflen = msg.n.nlmsg_len;
	memset(&nladdr, 0, sizeof(nladdr));
	nladdr.nl_family = AF_NETLINK;

	/* 循環發送直到發送完成 */
	while ((r = sendto(sd, buf, buflen, 0, (struct sockaddr *) &nladdr,
			   sizeof(nladdr))) < buflen) {
		if (r > 0) {
			buf += r;
			buflen -= r;
		} else if (errno != EAGAIN)
			return -1;
	}
	
	return 0;
}

消息的封裝過程同內核態消息封裝過程類似，需嚴格按照genelink消息格式進行封裝。首先填充netlink消息頭，其中nlmsg_type字段不使用netlink定義的標准type，填充為目標family的ID號，其他字段同其他類型的netlink類似；然後填充genetlink消息頭，這裡設定消息cmd字段為CTRL_CMD_GETFAMILY，version字段為DEMO_GENL_VERSION（同內核保持一致）；最後填充一個attr屬性，其中屬性頭的nla_type設定為函數傳入的屬性type，現該值為CTRL_ATTR_FAMILY_NAME，然後將傳入的family
name拷貝到屬性attr的payload載荷中，最後更新各個消息頭中的長度字段。
消息分裝完成後調用sendto系統調用啟動發送流程，指定目的地址的地址簇為AF_NETLINK，ID號為0（表示內核）。sendto函數同前博文《Netlink 內核實現分析（二）：通信》中分析的sendmsg()系統調用類似（sendto的msg消息封裝過程由內核完成），最後都是調用到sock_sendmsg()函數，具體的中間發送流程前博文中已詳細描述，這裡不再贅述，直接進入到發送的最後階段，來看Ctrl簇是如何處理接收到的查詢消息的。
在netlink函數調用流程的最後會調用具體協議類型的netlink_rcv()回調函數，其中genetlink的回調函數在前文中已經看到為genl_rcv()：

static void genl_rcv(struct sk_buff *skb)
{
	down_read(&cb_lock);
	netlink_rcv_skb(skb, &genl_rcv_msg);
	up_read(&cb_lock);
}

這裡netlink_rcv_skb函數的兩個入參其中第一個為消息skb，第二個為genl_rcv_msg回調函數；netlink_rcv_skb()函數會對消息進行一些通用性的處理，將用戶消息封裝成genl_info結構，最後會把消息控制權交給genl_rcv_msg()回調函數：

int netlink_rcv_skb(struct sk_buff *skb, int (*cb)(struct sk_buff *,
						     struct nlmsghdr *))
{
	struct nlmsghdr *nlh;
	int err;

	while (skb->len >= nlmsg_total_size(0)) {
		int msglen;

		nlh = nlmsg_hdr(skb);
		err = 0;

		if (nlh->nlmsg_len < NLMSG_HDRLEN || skb->len < nlh->nlmsg_len)
			return 0;

		/* Only requests are handled by the kernel */
		if (!(nlh->nlmsg_flags & NLM_F_REQUEST))
			goto ack;

		/* Skip control messages */
		if (nlh->nlmsg_type < NLMSG_MIN_TYPE)
			goto ack;

		err = cb(skb, nlh);
		if (err == -EINTR)
			goto skip;

ack:
		if (nlh->nlmsg_flags & NLM_F_ACK || err)
			netlink_ack(skb, nlh, err);

skip:
		msglen = NLMSG_ALIGN(nlh->nlmsg_len);
		if (msglen > skb->len)
			msglen = skb->len;
		skb_pull(skb, msglen);
	}

	return 0;
}

首先判斷消息的長度是否不小於netlink消息頭的長度（現在的上下文中顯然成立），然後進入while循環開始處理存放在skb中的netlink消息（可能有多個）。循環處理中會首先進行一些基本的數據長度判斷，然後根據nlmsg_flags和nlmsg_type字段判斷是否跳過消息處理流程、以及是否回發ACK相應。目前由於設定的nlmsg_flags為NLM_F_REQUEST、nlmsg_type為GENL_ID_CTRL（即NLMSG_MIN_TYPE），因此調用genl_rcv_msg()回調函數開始消息處理流程：

static int genl_rcv_msg(struct sk_buff *skb, struct nlmsghdr *nlh)
{
	struct genl_family *family;
	int err;

	family = genl_family_find_byid(nlh->nlmsg_type);
	if (family == NULL)
		return -ENOENT;

	if (!family->parallel_ops)
		genl_lock();

	err = genl_family_rcv_msg(family, skb, nlh);

	if (!family->parallel_ops)
		genl_unlock();

	return err;
}

該函數首先通過nlmsg_type字段（即family id號）在散列表中查找到對應的注冊family，然後如果消息處理不可重入，則這裡會上鎖，接下來調用genl_family_rcv_msg()函數：

static int genl_family_rcv_msg(struct genl_family *family,
			       struct sk_buff *skb,
			       struct nlmsghdr *nlh)
{
	const struct genl_ops *ops;
	struct net *net = sock_net(skb->sk);
	struct genl_info info;
	struct genlmsghdr *hdr = nlmsg_data(nlh);
	struct nlattr **attrbuf;
	int hdrlen, err;

	/* this family doesn't exist in this netns */
	if (!family->netnsok && !net_eq(net, &init_net))
		return -ENOENT;

	hdrlen = GENL_HDRLEN + family->hdrsize;
	if (nlh->nlmsg_len < nlmsg_msg_size(hdrlen))
		return -EINVAL;

函數首先判斷網絡命名空間，若不支持則當前消息的網絡空間必須為init_net，然後判斷消息的長度。

ops = genl_get_cmd(hdr->cmd, family);
	if (ops == NULL)
		return -EOPNOTSUPP;

static const struct genl_ops *genl_get_cmd(u8 cmd, struct genl_family *family)
{
	int i;

	for (i = 0; i < family->n_ops; i++)
		if (family->ops[i].cmd == cmd)
			return &family->ops[i];

	return NULL;
}

這裡找到消息cmd命令對應的處理函數並保存早ops變量中，查找的方式是通過cmd字段的匹配類型來找的，這裡找到的就是前文中注冊的demo_ops結構了。

if ((ops->flags & GENL_ADMIN_PERM) &&
	    !netlink_capable(skb, CAP_NET_ADMIN))
		return -EPERM;

接下來判斷權限，這裡由於已經在demo_ops中設置了GENL_ADMIN_PERM標識，因此本命令操作需要具有CAP_NET_ADMIN權限。

if ((nlh->nlmsg_flags & NLM_F_DUMP) == NLM_F_DUMP) {
		int rc;

		if (ops->dumpit == NULL)
			return -EOPNOTSUPP;

		if (!family->parallel_ops) {
			struct netlink_dump_control c = {
				.module = family->module,
				/* we have const, but the netlink API doesn't */
				.data = (void *)ops,
				.dump = genl_lock_dumpit,
				.done = genl_lock_done,
			};

			genl_unlock();
			rc = __netlink_dump_start(net->genl_sock, skb, nlh, &c);
			genl_lock();

		} else {
			struct netlink_dump_control c = {
				.module = family->module,
				.dump = ops->dumpit,
				.done = ops->done,
			};

			rc = __netlink_dump_start(net->genl_sock, skb, nlh, &c);
		}

		return rc;
	}

如果用戶設定了NLM_F_DUMP標識，這裡就會調用啟動dump流程，回填skb消息（這裡的skb將不再是用戶下發的消息了）。這裡不進行詳細的分析，繼續往下看：

if (family->maxattr && family->parallel_ops) {
		attrbuf = kmalloc((family->maxattr+1) *
					sizeof(struct nlattr *), GFP_KERNEL);
		if (attrbuf == NULL)
			return -ENOMEM;
	} else
		attrbuf = family->attrbuf;

這裡為attr屬性指定接收緩存，在支持重入的情況下這裡會另行動態分配內存，否則使用在注冊family的__genl_register_family函數中分配的內存空間。需要注意的是這裡的內存其實只是一個指針數組，用來存放attr屬性的地址，並不會存放實際的屬性數據。

if (attrbuf) {
		err = nlmsg_parse(nlh, hdrlen, attrbuf, family->maxattr,
				  ops->policy);
		if (err < 0)
			goto out;
	}

這裡將消息的數據拷貝到緩存空間中去，nlmsg_parse()的幾個入參分別為netlink消息頭，genelink消息頭長度（其實也包括了用戶私有頭，只不過這裡為0罷了），數據屬性緩存地址，緩存空間大小和屬性有效性策略結構。

static inline int nlmsg_parse(const struct nlmsghdr *nlh, int hdrlen,
			      struct nlattr *tb[], int maxtype,
			      const struct nla_policy *policy)
{
	if (nlh->nlmsg_len < nlmsg_msg_size(hdrlen))
		return -EINVAL;

	return nla_parse(tb, maxtype, nlmsg_attrdata(nlh, hdrlen),
			 nlmsg_attrlen(nlh, hdrlen), policy);
}

該函數間接調用netlink通用的屬性拷貝函數，其中將第三個參數為attr參數的首地址：nlmsg_attrdata(nlh, hdrlen)：

static inline struct nlattr *nlmsg_attrdata(const struct nlmsghdr *nlh,
					    int hdrlen)
{
	unsigned char *data = nlmsg_data(nlh);
	return (struct nlattr *) (data + NLMSG_ALIGN(hdrlen));
}

這裡將指針跳過netlink的頭以及genelink頭，指向attr的首地址。
第四個參數為attr屬性的長度：nlmsg_attrlen(nlh, hdrlen)：

static inline int nlmsg_attrlen(const struct nlmsghdr *nlh, int hdrlen)
{
	return nlmsg_len(nlh) - NLMSG_ALIGN(hdrlen);
}

計算方式為消息除去netlink消息頭的剩余長度減去genetlink消息頭長度後的長度。

int nla_parse(struct nlattr **tb, int maxtype, const struct nlattr *head,
	      int len, const struct nla_policy *policy)
{
	const struct nlattr *nla;
	int rem, err;

	memset(tb, 0, sizeof(struct nlattr *) * (maxtype + 1));

	nla_for_each_attr(nla, head, len, rem) {
		u16 type = nla_type(nla);

		if (type > 0 && type <= maxtype) {
			if (policy) {
				err = validate_nla(nla, maxtype, policy);
				if (err < 0)
					goto errout;
			}

			tb[type] = (struct nlattr *)nla;
		}
	}

	if (unlikely(rem > 0))
		pr_warn_ratelimited("netlink: %d bytes leftover after parsing attributes in process `%s'.\n",
				    rem, current->comm);

	err = 0;
errout:
	return err;
}

可以看到該函數會逐一的將屬性的地址復制到tb指針數組中去，但是如果傳入了有效性策略，那他就會調用validate_nla函數執行有效性判斷。對於這裡傳入的CTRL_ATTR_FAMILY_NAME屬性來說，在ctrl_policy中已經定義了有效性限制為NLA_NUL_STRING，最大長度為GENL_NAMSIZ-1：

static const struct nla_policy ctrl_policy[CTRL_ATTR_MAX+1] = {
	[CTRL_ATTR_FAMILY_ID]	= { .type = NLA_U16 },
	[CTRL_ATTR_FAMILY_NAME]	= { .type = NLA_NUL_STRING,
				    .len = GENL_NAMSIZ - 1 },
};

回到genl_family_rcv_msg()函數中繼續往下分析：

info.snd_seq = nlh->nlmsg_seq;
	info.snd_portid = NETLINK_CB(skb).portid;
	info.nlhdr = nlh;
	info.genlhdr = nlmsg_data(nlh);
	info.userhdr = nlmsg_data(nlh) + GENL_HDRLEN;
	info.attrs = attrbuf;
	info.dst_sk = skb->sk;
	genl_info_net_set(&info, net);
	memset(&info.user_ptr, 0, sizeof(info.user_ptr));

這裡開始封裝genl_info消息結構，填充對應的字段，比較好理解，其中snd_portid填充為發送端的套接字ID號，attrs為前文中分配的attr緩存空間首地址，接下來啟動最終的調用處理流程：

if (family->pre_doit) {
		err = family->pre_doit(ops, skb, &info);
		if (err)
			goto out;
	}

	err = ops->doit(skb, &info);

	if (family->post_doit)
		family->post_doit(ops, skb, &info);

如果在注冊family時指定了pre_doit和post_doit回調函數，將在分別調用ops->doit()函數的前後調用他們，對於Ctrl簇而言並沒有定義，這裡會直接調用ops->doit()回調函數，對於CTRL_CMD_GETFAMILY來說就是ctrl_getfamily()了：

static int ctrl_getfamily(struct sk_buff *skb, struct genl_info *info)
{
	struct sk_buff *msg;
	struct genl_family *res = NULL;
	int err = -EINVAL;

	if (info->attrs[CTRL_ATTR_FAMILY_ID]) {
		u16 id = nla_get_u16(info->attrs[CTRL_ATTR_FAMILY_ID]);
		res = genl_family_find_byid(id);
		err = -ENOENT;
	}

首先函數匹配CTRL_ATTR_FAMILY_ID，由於並未傳入該屬性數據，因此這裡該屬性的地址為NULL，然後接著判斷另一個屬性類型：

if (info->attrs[CTRL_ATTR_FAMILY_NAME]) {
		char *name;

		name = nla_data(info->attrs[CTRL_ATTR_FAMILY_NAME]);
		res = genl_family_find_byname(name);

		err = -ENOENT;
	}

這裡就開始處理CTRL_CMD_GETFAMILY屬性了，只做了一件事，就是通過用戶傳入的family name獲取到對應的family結構。

if (res == NULL)
		return err;

	if (!res->netnsok && !net_eq(genl_info_net(info), &init_net)) {
		/* family doesn't exist here */
		return -ENOENT;
	}

	msg = ctrl_build_family_msg(res, info->snd_portid, info->snd_seq,
				    CTRL_CMD_NEWFAMILY);
	if (IS_ERR(msg))
		return PTR_ERR(msg);

	return genlmsg_reply(msg, info);

這裡依然使用ctrl_build_family_msg()函數封裝回發消息（該函數的分析見前文），注意回發消息的cmd為CTRL_CMD_NEWFAMILY（它會將查詢結果family的全部內容回傳），指定的port_id號為消息查詢端的id（並不是內核的id號0），消息的sequence也同查詢消息一致。
函數最後調用genlmsg_reply()向應用層回發消息：

static inline int genlmsg_reply(struct sk_buff *skb, struct genl_info *info)
{
	return genlmsg_unicast(genl_info_net(info), skb, info->snd_portid);
}

可以看到它就是nlmsg_unicast的一個封裝而已（nlmsg_unicast的實現分析見《Netlink 內核實現分析（二）：通信》）。至此查詢消息的發送和內核的處理流程分析完畢，下面回到示例程序demo_get_family_id()中：

/* 接收內核消息 */
	rep_len = recv(sd, &ans, sizeof(ans), 0);
	if (ans.n.nlmsg_type == NLMSG_ERROR || (rep_len < 0) || !NLMSG_OK((&ans.n), rep_len))
		return 0;

	/* 解析family id */
	na = (struct nlattr *) GENLMSG_DATA(&ans);
	na = (struct nlattr *) ((char *) na + NLA_ALIGN(na->nla_len));
	if (na->nla_type == CTRL_ATTR_FAMILY_ID) {
		id = *(__u16 *) NLA_DATA(na);
	}

這裡找到回發消息中的第二個attr（消息結構參見圖1-c），然後獲取出其中的family id號。至此用戶程序成功獲取的了demo family的id號，接下來就可以向他發送消息了。

2、向內核Demo Family發送消息

/* 發送字符串消息 */
	my_pid = getpid();
	string = argv[1];
	data = atoi(argv[2]);
	
	ret = demo_send_cmd(nl_fd, nl_family_id, my_pid, DEMO_CMD_ECHO,
			  DEMO_CMD_ATTR_MESG, string, strlen(string) + 1);
	if (ret < 0) {
		fprintf(stderr, "failed to send echo cmd\n");
		goto out;
	}

	/* 發送數據消息 */
	ret = demo_send_cmd(nl_fd, nl_family_id, my_pid, DEMO_CMD_ECHO,
			  DEMO_CMD_ATTR_DATA, &data, sizeof(data));
	if (ret < 0) {
		fprintf(stderr, "failed to send echo cmd\n");
		goto out;
	}

本示例程序比較簡單，直接使用程序的入參作為發送的數據。發送依然是調用demo_send_cmd函數實現，但是入參同獲取family id時的有所不同，首先發送字符串消息時第二個入參設置為剛剛獲取的demo family id，然後發送端套接字ID為當前進程的pid號，然後發送cmd為DEMO_CMD_ECHO，發送的屬性依次為DEMO_CMD_ATTR_DATA（其實cmd和attr屬性並沒有明確的一一對應關系，用戶可根據需求自行組合，同時一個cmd消息也可以帶很多的attr屬性，這點從內核ctrl回發的消息就可以看出），最後發送的消息內容分別為用戶輸入的字符串和數據。

3、內核Demo Family 回發消息

應用層序向demo family發送DEMO_CMD_ECHO消息後，內核會調用到前文中注冊時指定的doit回調函數demo_echo_cmd（具體的數據發送流程同前文中分析的Ctrl查詢消息，不再詳細分析），來看一下demo_echo_cmd()函數所做的處理。

static int demo_echo_cmd(struct sk_buff *skb, struct genl_info *info)
{
	if (info->attrs[DEMO_CMD_ATTR_MESG])
		return cmd_attr_echo_message(info);
	else if (info->attrs[DEMO_CMD_ATTR_DATA])
		return cmd_attr_echo_data(info);
	else
		return -EINVAL;
}

該函數會判斷接收的屬性類型，並做出相應的處理（注意：為了簡單起見，該doit回調函數最多一次只能處理一種類型的attr屬性），先來看cmd_attr_echo_message()函數

static int cmd_attr_echo_message(struct genl_info *info)
{
	struct nlattr *na; 
	char *msg;  
	struct sk_buff *rep_skb;
	size_t size;
	int ret;

	/* 讀取用戶下發的消息 */
	na = info->attrs[DEMO_CMD_ATTR_MESG];
	if (!na)
		return -EINVAL;

	msg = (char *)nla_data(na);
	pr_info("demo generic netlink receive echo mesg %s\n", msg);  

	/* 回發消息 */
	size = nla_total_size(strlen(msg)+1);
	
	/* 准備構建消息 */
	ret = demo_prepare_reply(info, DEMO_CMD_REPLY, &rep_skb, size);
	if (ret < 0)
		return ret;

	/* 填充消息 */
	ret = demo_mk_reply(rep_skb, DEMO_CMD_ATTR_MESG, msg, size); 
	if (ret < 0)
		goto err;

	/* 完成構建並發送 */
	return demo_send_reply(rep_skb, info);

err:
	nlmsg_free(rep_skb);
	return ret;	
}

這裡從DEMO_CMD_ATTR_MESG屬性地址處取出用戶下發的消息內容，然後調用demo_prepare_reply構建回發消息頭。其中入參依次為接收genl_info消息，回發cmd類型，skb指針地址，回發數據長度。

static int demo_prepare_reply(struct genl_info *info, u8 cmd, struct sk_buff **skbp, size_t size)
{
	struct sk_buff *skb;
	void *reply;

	/*
	 * If new attributes are added, please revisit this allocation
	 */
	skb = genlmsg_new(size, GFP_KERNEL);
	if (!skb)
		return -ENOMEM;

	if (!info)
		return -EINVAL;

	/* 構建回發消息頭 */	
	reply = genlmsg_put_reply(skb, info, &demo_family, 0, cmd);
	if (reply == NULL) {
		nlmsg_free(skb);
		return -EINVAL;
	}

	*skbp = skb;
	return 0;
}

這裡依然調用genlmsg_new()函數申請skb套接字緩存空間，然後直接調用genlmsg_put_reply()函數構建回發消息的netlink消息頭和genetlink消息頭：

static inline void *genlmsg_put_reply(struct sk_buff *skb,
				      struct genl_info *info,
				      struct genl_family *family,
				      int flags, u8 cmd)
{
	return genlmsg_put(skb, info->snd_portid, info->snd_seq, family,
			   flags, cmd);
}

該函數僅僅是genlmsg_put的一個封裝而已，注意入參info->snd_portid為用戶層的netlink套接字的id號。回到cmd_attr_echo_message()函數中，接下來填充消息屬性：

/* 填充消息 */
	ret = demo_mk_reply(rep_skb, DEMO_CMD_ATTR_MESG, msg, size); 
	if (ret < 0)
		goto err;

static int demo_mk_reply(struct sk_buff *skb, int aggr, void *data, int len)  
{  
    /* add a netlink attribute to a socket buffer */  
    return nla_put(skb, aggr, len, data); 
}

這裡調用nla_put()函數將字符串消息填充到第一個attr屬性中，同時指定attr的屬性類型為DEMO_CMD_ATTR_MESG，最後調用demo_send_reply()將消息往應用層發送：

/* 完成構建並發送 */
	return demo_send_reply(rep_skb, info);

static int demo_send_reply(struct sk_buff *skb, struct genl_info *info)
{
	struct genlmsghdr *genlhdr = nlmsg_data(nlmsg_hdr(skb));
	void *reply = genlmsg_data(genlhdr);

	genlmsg_end(skb, reply);

	return genlmsg_reply(skb, info);
}

首先調用genlmsg_end()更新消息頭重的長度字段，然後調用genlmsg_reply啟動回發流程：

static inline int genlmsg_reply(struct sk_buff *skb, struct genl_info *info)
{
	return genlmsg_unicast(genl_info_net(info), skb, info->snd_portid);
}

該函數為genlmsg_unicast()的一個封裝。這樣demo family的回發字符串息就發送出去了。下面再來簡單的看一下cmd_attr_echo_data()函數，它同cmd_attr_echo_message()函數基本類似，唯一的區別就是調用了內核提供的nla_get_s32()和nla_put_s32()這兩個封裝函數來獲取和設置s32類型的attr屬性，不做過多的論描。

static int cmd_attr_echo_data(struct genl_info *info)
{
	struct nlattr *na; 
	s32	data;
	struct sk_buff *rep_skb;
	size_t size;
	int ret;

	/* 讀取用戶下發的數據 */
	na = info->attrs[DEMO_CMD_ATTR_DATA];
	if (!na)
		return -EINVAL;

	data = nla_get_s32(info->attrs[DEMO_CMD_ATTR_DATA]);
	pr_info("demo generic netlink receive echo data %d\n", data);  

	/* 回發數據 */
	size = nla_total_size(sizeof(s32));

	ret = demo_prepare_reply(info, DEMO_CMD_REPLY, &rep_skb, size);
	if (ret < 0)
		return ret;

	/* 為了簡單這裡直接調用netlink庫函數(對於需求的豐富可以自行封裝) */
	ret = nla_put_s32(rep_skb, DEMO_CMD_ATTR_DATA, data);
	if (ret < 0)
		goto err;

	return demo_send_reply(rep_skb, info);

err:
	nlmsg_free(rep_skb);
	return ret; 
}

內核消息全部單播發送出去以後，下面來看應用層的接收流程。

4、應用層接收內核Demo Family回發消息

int main(int argc, char* argv[]) 
{
	......

	/* 接收用戶消息並解析(本示例程序中僅解析2個) */
	demo_msg_recv_analysis(nl_fd, argc-1);

	......
}

void demo_msg_recv_analysis(int sd, int num)
{
	int rep_len;
	int len;
	struct nlattr *na;	
	struct msgtemplate msg;
	
	unsigned int data;
	char *string;

	while (num--) {
		
		/* 接收內核消息回顯 */
		rep_len = recv(sd, &msg, sizeof(msg), 0);
		if (rep_len < 0 || demo_msg_check(msg, rep_len) < 0) {
			fprintf(stderr, "nonfatal reply error: errno %d\n", errno);
			continue;
		}
		PRINTF("received %d bytes\n", rep_len);
		PRINTF("nlmsghdr size=%zu, nlmsg_len=%d, rep_len=%d\n",
			   sizeof(struct nlmsghdr), msg.n.nlmsg_len, rep_len);
		
		rep_len = GENLMSG_PAYLOAD(&msg.n);
		na = (struct nlattr *) GENLMSG_DATA(&msg);
		len = 0;
		
		/* 一個msg裡可能有多個attr，所以這裡循環讀取 */
		while (len < rep_len) {
			len += NLA_ALIGN(na->nla_len);
			switch (na->nla_type) {
			case DEMO_CMD_ATTR_MESG:
				/* 接收到內核字符串回顯 */
				string = (char *) NLA_DATA(na);
				printf("echo reply:%s\n", string);
				break;
			case DEMO_CMD_ATTR_DATA:
				/* 接收到內核數據回顯 */
				data = *(int *) NLA_DATA(na);
				printf("echo reply:%u\n", data);
				break;	
			default:
				fprintf(stderr, "Unknown nla_type %d\n", na->nla_type);
			}
			na = (struct nlattr *) (GENLMSG_DATA(&msg) + len);
		}	
	}
}

應用程序調用demo_msg_recv_analysis()函數接收內核消息，其中num表示接收消息的個數。該函數中循環調用recv函數阻塞式的接收內核netlink消息。當有消息接收到以後調用demo_msg_check()函數判斷消息的有效性：

int demo_msg_check(struct msgtemplate msg, int rep_len)
{
	if (msg.n.nlmsg_type == NLMSG_ERROR || !NLMSG_OK((&msg.n), rep_len)) {
		struct nlmsgerr *err = NLMSG_DATA(&msg);
		fprintf(stderr, "fatal reply error,  errno %d\n", err->error);
		return -1;
	}
	
	return 0;
}

這裡首先判斷消息頭中的nlmsg_type字段，如果該字段為NLMSG_ERROR表示接收到了錯誤的消息，應該立即丟棄。如果接收到的消息類型無誤，則接下來判斷消息的長度是否足夠，使用的是NLMSG_OK宏（見netlink.h）。然後接收函數循環讀取attr屬性並根據屬性的attr類型單獨進行處理，本示例中僅僅在終端中打印。需要補充的是，本程序中並沒有對接收到的消息cmd類型進行判斷，其實為了程序的可靠性考慮，最好增加這一方面的判斷（雖然netlink的id號保證了不會收到其他id的genetlink消息，但是當某family的cmd類型較多時容易引起混亂）。至此demo
family的genetlink單播通信過程就大致分析完畢，另外關於多播的通信流程也比較簡單，不難掌握，以後有時間再行補充。

四、總結

本文和前一篇《Generic Netlink內核實現分析（一）：初始化》分析了內核與用戶態Generic netlink的創建、創建及通信流程。Generic netlink作為內核netlink機制的一種特殊應用，更好的實現了內核態與用戶態之間的全雙工通信的拓展，同時本文還給出了一個簡單的示例程序用以分析原理，亦可用作日後擴展的原型程序。
參考文獻：1、《Linux Kernel Networking Implementation and Theory》；2、《Generic Netlink詳解》