考虑下面这个命令：
who > userlist
当这个命令完成后，文件系统增加l一个存放命令who输出内容的新文件。

	
内核之解析DTS设备树文件并创建设备的过程
	
在这里,我分析的是内核源码来自谷歌官方Android7.1.2源码包经过打补丁包"SC60_Android7.1.2_Quectel_SDK_r270060_20180731.tar.gz"后得到的.

本文分析时使用的工具是"SourceInsight".

分析过程中抓大放小,仅分析主干流程,细枝末节的东西未深究,如果有同志愿意分享自己分析的心得,或者进行了更深层次的分析,还望贴出自己博客链接在评论栏,或者把相关资料发到我的邮箱"uestc_ganlin@163.com",谢谢~


	
主干的分析的过程
	
搜索有关宏定义DT_MACHINE_START的所有文件发现:该宏定义是在"kernel\msm-3.18\arch\arm\include\asm\mach"目录的"arch.h",除此之外,其他文件都为这个宏定义的使用文件,形如在目录"kernel\msm-3.18\arch\arm\mach-xxx"下,名为"xxx.c"的文件.这里我用的是"kernel\msm-3.18\arch\arm\mach-msm\board-8953.c"来切入,进行后续分析:

DT_MACHINE_START(MSM8953_DT,
	"Qualcomm Technologies, Inc. MSM 8953 (Flattened Device Tree)")
	.init_machine = msm8953_init,
	.dt_compat = msm8953_dt_match,
MACHINE_END


发现其实主要做了两件是,一个是定义msm8953的设备树匹配表dt_compat为msm8953_dt_match,另一个是定义msm8953的初始化函数init_machine为msm8953_init.

先来看看msm8953的设备树匹配表msm8953_dt_match:

static const char *msm8953_dt_match[] __initconst = {
	"qcom,msm8953",
	"qcom,apq8053",
	NULL
};

发现msm8953_dt_match是一个char*类型的指针数组,其中每个元素指向一个字符串,譬如"qcom,msm8953".

这个字符串我猜测的是用来指明要匹配的DTS文件的,因为所有的DTS文件都在"kernel\msm-3.18\arch\arm\boot\dts\"目录下,所以这里指明的相关文件应该就是"kernel\msm-3.18\arch\arm\boot\dts\qcom"目录下的"msm8953.dtsi"文件;

或者查看该DTS文件中根节点定义的compatible属性,发现该属性定义为 "qcom,msm8953",看到这里,也可能因为他跟该DTS文件中定义的compatible属性一致,所以才能匹配的上.但不论是前者那种我自己猜的理解方式("qcom,msm8953"是指明目录和文件的),还是后者的理解("qcom,msm8953"是通过DTS文件中定义的compatible属性进行匹配的)都能够说明该字符串的用意,貌似后者的理解更为合理一些,呵呵~

接下来继续分析msm8953的初始化函数msm8953_init:

static void __init msm8953_init(void)
{
	board_dt_populate(NULL);
}

简单分析后不难看出msm8953_init是初始化设备的入口函数,而 msm8953_dt_match是设备树匹配的关键数据结构数组.

继续看msm8953_init函数中调用的板级设备树建立函数board_dt_populate,切换到定义处所在文件"kernel\msm-3.18\arch\arm\mach-msm\board-dt.c":

void __init board_dt_populate(struct of_dev_auxdata *adata)
{
	of_platform_populate(NULL, of_default_bus_match_table, NULL, NULL);

	/* Explicitly parent the /soc devices to the root node to preserve
	 * the kernel ABI (sysfs structure, etc) until userspace is updated
	 */
	of_platform_populate(of_find_node_by_path("/soc"),
			     of_default_bus_match_table, adata, NULL);
}


其中两个关键点,一个该函数的参数是of_device_id结构体类型的数组定义的默认总线匹配表of_default_bus_match_table:

const struct of_device_id of_default_bus_match_table[] = {
	{ .compatible = "simple-bus", },
	{ .compatible = "simple-mfd", },
#ifdef CONFIG_ARM_AMBA
	{ .compatible = "arm,amba-bus", },
#endif /* CONFIG_ARM_AMBA */
	{} /* Empty terminated list */
};

如代码所示,这是一个of_device_id结构体类型的数组,其中为每一个元素的compatible赋值,不难理解,这个就真的是会和DTS中对应的compatible属性进行匹配了.

另一个关键点是该函数调用的平台设备建立函数of_platform_populate,在文件"kernel\msm-3.18\drivers\of\platform.c":

int of_platform_populate(struct device_node *root,
			const struct of_device_id *matches,
			const struct of_dev_auxdata *lookup,
			struct device *parent)
{
	struct device_node *child;
	int rc = 0;

	root = root ? of_node_get(root) : of_find_node_by_path("/");
	if (!root)
		return -EINVAL;

	for_each_child_of_node(root, child) {
		rc = of_platform_bus_create(child, matches, lookup, parent, true);
		if (rc)
			break;
	}
	of_node_set_flag(root, OF_POPULATED_BUS);

	of_node_put(root);
	return rc;
}

 核心工作是用for_each_child_of_node函数遍历根设备节点下的所有的子节点,并执行of_platform_bus_create函数对每一个子节点进行平台总线设备创建:

static int of_platform_bus_create(struct device_node *bus,
				  const struct of_device_id *matches,
				  const struct of_dev_auxdata *lookup,
				  struct device *parent, bool strict)
{
	const struct of_dev_auxdata *auxdata;
	struct device_node *child;
	struct platform_device *dev;
	const char *bus_id = NULL;
	void *platform_data = NULL;
	int rc = 0;

	/* Make sure it has a compatible property */
	if (strict && (!of_get_property(bus, "compatible", NULL))) {
		pr_debug("%s() - skipping %s, no compatible prop\n",
			 __func__, bus->full_name);
		return 0;
	}

	auxdata = of_dev_lookup(lookup, bus);
	if (auxdata) {
		bus_id = auxdata->name;
		platform_data = auxdata->platform_data;
	}

	if (of_device_is_compatible(bus, "arm,primecell")) {
		/*
		 * Don't return an error here to keep compatibility with older
		 * device tree files.
		 */
		of_amba_device_create(bus, bus_id, platform_data, parent);
		return 0;
	}

	dev = of_platform_device_create_pdata(bus, bus_id, platform_data, parent);
	if (!dev || !of_match_node(matches, bus))
		return 0;

	for_each_child_of_node(bus, child) {
		pr_debug("   create child: %s\n", child->full_name);
		rc = of_platform_bus_create(child, matches, lookup, &dev->dev, strict);
		if (rc) {
			of_node_put(child);
			break;
		}
	}
	of_node_set_flag(bus, OF_POPULATED_BUS);
	return rc;
}


平台总线设备创建的主要工作流程是,先通过of_get_property函数检查该平台总线设备节点否具有compatible属性,通过后开始执行of_platform_device_create_pdata函数真正的创建平台设备内容,到这里其实就已经算是完成了一个设备节点从匹配识别到最终创建出设备内容的完整流程了,但是了解DTC文件的朋友知道,这个设备节点下往往还会有其他的设备的子节点,然后子节点的子节点还有设备节点......等等,以此类推组成的一个庞大的树形分支的设备树结构,那怎么办得呢?往下看就知道了,其实很简单,管它有多少个设备的节点,都调用到我们这里把设备内容真正的创建出来不就解决了吗?是的,的确如此,接下来再使用for_each_child_of_node函数遍历当前平台总线设备节点的所有子节点,递归式的执行of_platform_bus_create函数进行每一个平台总线设备的创建,其中每完成一个创建,调用of_node_put函数释放当前节点的内存.分析到这里,我对递归的用法算是理解的更深刻一点了.

接下来在看看具体如何创建平台设备内容的of_platform_device_create_pdata函数:

static struct platform_device *of_platform_device_create_pdata(
					struct device_node *np,
					const char *bus_id,
					void *platform_data,
					struct device *parent)
{
	struct platform_device *dev;

	if (!of_device_is_available(np) ||
	    of_node_test_and_set_flag(np, OF_POPULATED))
		return NULL;

	dev = of_device_alloc(np, bus_id, parent);
	if (!dev)
		goto err_clear_flag;

	dev->dev.bus = &platform_bus_type;
	dev->dev.platform_data = platform_data;
	of_dma_configure(&dev->dev, dev->dev.of_node);
	of_msi_configure(&dev->dev, dev->dev.of_node);
	of_reserved_mem_device_init(&dev->dev);

	if (of_device_add(dev) != 0) {
		of_reserved_mem_device_release(&dev->dev);
		of_dma_deconfigure(&dev->dev);
		platform_device_put(dev);
		goto err_clear_flag;
	}

	return dev;

err_clear_flag:
	of_node_clear_flag(np, OF_POPULATED);
	return NULL;
}

其中主要是,通过of_device_is_available函数检查设备节点是否有效,然后执行of_device_alloc函数分配设备内存,有了设备后,紧接着对设备的一部分内容进行初始化巴拉巴拉,最终调用of_device_add函数往内核中添加设备.

再看看是如何分配设备内容的of_device_alloc函数:

struct platform_device *of_device_alloc(struct device_node *np,
				  const char *bus_id,
				  struct device *parent)
{
	struct platform_device *dev;
	int rc, i, num_reg = 0, num_irq;
	struct resource *res, temp_res;

	dev = platform_device_alloc("", -1);
	if (!dev)
		return NULL;

	/* count the io and irq resources */
	while (of_address_to_resource(np, num_reg, &temp_res) == 0)
		num_reg++;
	num_irq = of_irq_count(np);

	/* Populate the resource table */
	if (num_irq || num_reg) {
		res = kzalloc(sizeof(*res) * (num_irq + num_reg), GFP_KERNEL);
		if (!res) {
			platform_device_put(dev);
			return NULL;
		}

		dev->num_resources = num_reg + num_irq;
		dev->resource = res;
		for (i = 0; i < num_reg; i++, res++) {
			rc = of_address_to_resource(np, i, res);
			WARN_ON(rc);
		}
		if (of_irq_to_resource_table(np, res, num_irq) != num_irq)
			pr_debug("not all legacy IRQ resources mapped for %s\n",
				 np->name);
	}

	dev->dev.of_node = of_node_get(np);
	dev->dev.parent = parent;

	if (bus_id)
		dev_set_name(&dev->dev, "%s", bus_id);
	else
		of_device_make_bus_id(&dev->dev);

	return dev;
}

首先调用platform_device_alloc函数为定义的platform_device结构体类型的设备的分配内存,接下来分别调用of_address_to_resource和of_irq_count两个函数统计该设备使用的地址和中断资源个数,如果有上述资源的存在则分别调用of_address_to_resource和of_irq_to_resource_table两个函数对设备进行地址和中断资源分配,完成上述设备初始化工作后紧接着更进一步初始化,比如调用of_node_get函数初始化设备所属设备树节点,最终执行dev_set_name或者of_device_make_bus_id函数来设置设备的名字或者设置设备总线ID.

到此为止,所有主干内容分析完毕,如果你把所有的标红的内容整理出来,你会知道此次分析的源码所做的核心事件是什么,只要是我自己亲自进行的代码分析,我都会用心的记录下来,愿同行小伙伴自己做源码分析时也这样记录,并分享链接到评论,大家互相学习,谢谢~

        本文是sockfs文件系统分析的第三篇文章，本文主要分析socket fd的创建过程，主要介绍系统调用

socket接口的处理流程，主要包括struct socket类型变量的创建、struct sock类型变量的创建以及这些变量与vfs的关联以及与具体的协议处理接口的关联等。因我们之前在第二篇文章《LINUX 套接字文件系统（sockfs）分析之二 相关结构体分析》中已经介绍了socketfs相关的结构体以及它们之间的关联，那针对socket的创建的分析，相对来说就简单许多。

     对于网络协议簇而言，包括ax25协议簇、inet4协议簇、inet6协议簇、ipx协议簇等，而每一个协议

簇中又包含多个协议，以inet4协议簇为例，又包含ipv4、tcp、udp、icmp等协议。针对网络协议簇与网络协议之间的关系，sockfs的架构也按此进行了划分。此处以socket系统调用为例（以inet4为例进行介绍）：

 

     在socket系统调用中，主要包括网络部分相关的初始化、socket与vfs关联设置这两部分，下面是socket的流程图，其主要调用两个函数完成socket的创建工作：

1.调用sock_create，完成网络相关的socket创建、inode创建、相关协议的处理接口函数指针的设置等；

2.sock_map_fd，申请文件fd以及文件描述符创建，并完成与socket的关联。



socket函数的代码如下，我们接着就以socket_create、sock_map_fd这两个接口进行分析。

SYSCALL_DEFINE3(socket, int, family, int, type, int, protocol)

{

int retval;

struct socket *sock;

int flags;



/* Check the SOCK_* constants for consistency.  */

BUILD_BUG_ON(SOCK_CLOEXEC != O_CLOEXEC);

BUILD_BUG_ON((SOCK_MAX | SOCK_TYPE_MASK) != SOCK_TYPE_MASK);

BUILD_BUG_ON(SOCK_CLOEXEC & SOCK_TYPE_MASK);

BUILD_BUG_ON(SOCK_NONBLOCK & SOCK_TYPE_MASK);



flags = type & ~SOCK_TYPE_MASK;

if (flags & ~(SOCK_CLOEXEC | SOCK_NONBLOCK))

return -EINVAL;

type &= SOCK_TYPE_MASK;



if (SOCK_NONBLOCK != O_NONBLOCK && (flags & SOCK_NONBLOCK))

flags = (flags & ~SOCK_NONBLOCK) | O_NONBLOCK;



retval = sock_create(family, type, protocol, &sock);

if (retval < 0)

goto out;



retval = sock_map_fd(sock, flags & (O_CLOEXEC | O_NONBLOCK));

if (retval < 0)

goto out_release;



out:

/* It may be already another descriptor 8) Not kernel problem. */

return retval;



out_release:

sock_release(sock);

return retval;

}

 

 

 

socket_create接口分析

socket_create接口是通过调用__sock_create接口进行socket创建与设置的，我们重点关注这个接口。

 

下面是__sock_create接口的处理流程图，其主要完成如下几个功能：

调用sock_alloc接口进行struct socket类型变量的创建、inode节点的创建，主要是通过调用sockfs的超级块的alloc_inode接口sock_alloc_inode接口，该接口创建struct socket_alloc类型的内存变量，该变量中包含了inode成员、socket成员，并完成inode节点的初始化与设置。针对sock_alloc_inode接口，我们之前在第一篇文章《LINUX 套接字文件系统（sockfs）分析之一文件系统注册分析》中已经介绍过，需要了解的查看这篇文章的内容。
	
根据传递的网络协议簇的类型（AF_INET、AF_INET6、AF_NETLINK等），从全局数组net_families中查找对应协议簇的struct net_proto_family类型的指针pf；
	
调用pf->create接口，继续进行socket的设置操作。



 

 

以上即为sock_create的处理流程，针对上述第3步的pf->create接口，进行进一步的初始化操作，主要就是进行具体协议的初始化以及该协议簇相关的变量以及接口的初始化操作。为了更好理解协议簇相关的create接口，我们以ipv4的create接口进行详细的分析与说明。

 

inet的create接口（inet_create）

     以上即为sock_create的处理流程，针对上述第3步的pf->create接口，此处以inet协议簇的create接口进行解析，inet的create接口名称为inet_create。该接口的处理流程如下：

1.将socket的状态社社长为SS_UNCONNETED；

2.根据socket类型、协议号，查找对应的struct inet_protosw类型的inet协议变量，若没有查找到，则返回创建失败；若成功，则继续执行如下步骤；

3.创建inet sock变量，并进行相应的初始化，包括sk_family、sk_proto、sk_net等成员的设置，同时inet sock中包含struct sock类型的成员；

4.设置socket变量的ops指针，ops为相应socket（SOCK_STREAM、SOCKET_DGRAM）类型对应的操作处理接口；

5.若sk->sk_proto->init指针不为空，则调用该接口进行具体协议的socket初始化操作（如针对tcp协议而言，则调用tcp_v4_init_sock接口进行相应的初始化）。

针对该接口涉及的结构体变量struct proto、struct proto_ops以及全局变量inetsw，之前的文章中已经介绍，此处不再赘述。

 



 

以上即为sock_create接口的处理流程，简而言之，即完成socket的创建、inode的创建、具体协议处理接口的设置等。

 

sock_map_fd接口分析

 

该接口主要完成文件描述符的创建以及文件描述符与socket的关联等信息，下面我们详细分析下

sock_map_fd接口的处理流程。

该接口实现的具体功能如下：

1.通过调用get_unused_fd_flags获取一个未使用的fd（关于get_unused_fd_flags，参见《LINUX 文件系统分析sys_open调用之get_unused_fd_flags接口分析》）；

2.调用sock_alloc_file接口，进行文件描述符的创建，并与socket进行关联；

3.调用fd_install接口实现文件描述符与文件fd的关联（关于fd_install相关的内容，参见《LINUX VFS分析之do_sys_open接口实现的简要说明》）。

 

以上即为sock_map_fd接口实现的主要功能。

 



 

sock_alloc_file接口分析

该接口实现的具体功能在上面流程图中已经说明，此处再说明一下：

A.调用d_alloc_pseudo创建socket对应的dentry；

B.调用alloc_file，创建文件描述符，并设置文件描述符的操作处理接口指针f_op指针为socket_file_ops；

而socket_file_ops的定义如下，其实现了aio_read、aio_write接口，而在系统调用read、write接口

的分析中（见《LINUX VFS分析之write调用接口的内核实现分析》、《LINUX VFS分析之read调用接口的内核实现分析》）可知，其会通过调用文件描述的f_op->read/f_op->aio_read、f_op->write/f_op->aio_write，实现文件的读写操作。因此针对socket而言，也可以使用read、write接口实现数据的收发操作。



 

static const struct file_operations socket_file_ops = {

.owner =        THIS_MODULE,

.llseek =        no_llseek,

.aio_read =        sock_aio_read,

.aio_write =        sock_aio_write,

.poll =                sock_poll,

.unlocked_ioctl = sock_ioctl,

#ifdef CONFIG_COMPAT

.compat_ioctl = compat_sock_ioctl,

#endif

.mmap =                sock_mmap,

.open =                sock_no_open,        /* special open code to disallow open via /proc */

.release =        sock_close,

.fasync =        sock_fasync,

.sendpage =        sock_sendpage,

.splice_write = generic_splice_sendpage,

.splice_read =        sock_splice_read,

};



 

 

至此我们完成了socket系统调用的实现流程。

 

【Hadoop】HDFS - 创建文件流程详解
转载自：foreach_break



本文目录 [+点此展开]

1.本文目的

通过解析客户端创建文件流程，认知hadoop的HDFS系统的一些功能和概念。
2.主要概念
2.1 NameNode（NN）：
HDFS系统核心组件，负责分布式文件系统的名字空间管理、INode表的文件映射管理。如果不开启备份/故障恢复/Federation模式，一般的HDFS系统就只有1个NameNode，当然这样是存在单点故障隐患的。
NN管理两个核心的表：文件到块序列的映射、块到机器序列的映射。

第一个表存储在磁盘中，第二表在NN每次启动后重建。
2.2 NameNodeServer（NNS）：
负责NN和其它组件的通信接口的开放（IPC、http）等。

NN通过客户端协议（ClientProtocol）和客户端通信，通过数据节点协议（DataNodeProtocol）和DN通信。
2.3 FSNameSystem：
管理文件系统相关，承担了NN的主要职责。
2.4 DataNode（DN）：
分布式文件系统中存放实际数据的节点，存储了一系列的文件块，一个DFS部署中通常有许多DN。

DN和NN，DN和DN，DN和客户端都通过不同的IPC协议进行交互。

通常，DN接受来自NN的指令，比如拷贝、删除文件块。

客户端在通过NN获取了文件块的位置信息后，就可以和DN直接交互，比如读取块、写入块数据等。
DN节点只管理一个核心表：文件块到字节流的映射。
在DN的生命周期中，不断地和NN通信，报告自己所存储的文件块的状态，NN不直接向DN通信，而是应答DN的请求，比如在DN的心跳请求后，回复一些关于复制、删除、恢复文件块的命令（comands）。

DN和外界通信的接口<host:port>会报告给NN，想和此DN交互的客户端或其它DN可以通过和NN通信来获取这一信息。
2.5 Block
文件块，hadoop文件系统的原语，hadoop分布式文件系统中存储的最小单位。一个hadoop文件就是由一系列分散在不同的DataNode上的block组成。
2.6 BlockLocation
文件块在分布式网络中的位置<host:port>，也包括一些块的元数据，比如块是否损坏、块的大小、块在文件中的偏移等。
2.7 DFSClient
分布式文件系统的客户端，用户可以获取一个客户端实例和NameNode及DataNode交互，DFSClient通过客户端协议和hadoop文件系统交互。
2.8 Lease
租约，当客户端创建或打开一个文件并准备进行写操作，NameNode会维护一个文件租约，以标记谁正在对此文件进行写操作。客户端需要定时更新租约，否则当租约过期，NN会关闭文件或者将文件的租约交给其它客户端。
2.9 LeaseRenewer
续约管控线程，当一个DFSClient调用申请租约后，如果此线程尚未启动，则启动，并定期向NameNode续约。
三.创建一个文件
当hadoop的分布式集群启动之后，可以通过FS或Shell来创建文件，FS创建文件的命令如下：

//cluser是hadoop集群,通过fs和集群文件系统交互
final DistributedFileSystem fs = cluster.getFileSystem();
// 要创建的文件名
final Path tmpFile1 = new Path("/tmpfile1.dat");

//创建文件
public static void createFile(FileSystem fs, Path fileName, long fileLen, 
      short replFactor, long seed) throws IOException {
    if (!fs.mkdirs(fileName.getParent())) {
      throw new IOException("Mkdirs failed to create " + 
                            fileName.getParent().toString());
    }
    FSDataOutputStream out = null;
    try {
      out = fs.create(fileName, replFactor);
      byte[] toWrite = new byte[1024];
      Random rb = new Random(seed);
      long bytesToWrite = fileLen;
      while (bytesToWrite>0) {
        rb.nextBytes(toWrite);
        int bytesToWriteNext = (1024<bytesToWrite)?1024:(int)bytesToWrite;

        out.write(toWrite, 0, bytesToWriteNext);
        bytesToWrite -= bytesToWriteNext;
      }
      out.close();
      out = null;
    } finally {
      IOUtils.closeStream(out);
    }
  }

四、流程分析
创建一个名为tmpfile1.dat的文件，主要流程如下：

4.1 发送创建文件的请求（CreateFile）
客户端向NN发起请求，获取文件信息，NN会在缓存中查找是否存在请求创建的文件项（file entry），如果没找到，就在NameSystem中创建一个新的文件项：

块管理器（BlockManager）检查复制因子是否在范围内，如果复制因子过小或过大就会异常。

同时会进行权限验证、加密、安全模式检测（如果在安全模式不能创建文件）等，并记录操作日志和事件日志，然后向客户端返回文件状态。

4.2 申请文件租用权（beginFileLease）
客户端取得文件状态后，对文件申请租用（lease），如果租用过期，客户端将无法再继续对文件进行访问，除非进行续租。
4.3 数据流管控线程启动（DataStreamer & ResponseProcessor）
DataStreamer线程负责数据的实际发送：

当数据队列（Data Queue）为空时，会睡眠，并定期苏醒以检测数据队列是否有新的数据需要发送、Socket套接字是否超时、是否继续睡眠等状态。

ResponseProcessor负责接收和处理pipeline下游传回的数据接收确认信息pipelineACK。
4.4 发送添加块申请并初始化数据管道（AddBlock & Setup Pipeline）
当有新的数据需要发送，并且块创建阶段处于PIPELINE_SETUP_CREATE，DataStreamer会和NameNode通信，调用AddBlock方法，通知NN创建、分配新的块及位置，NN返回后，初始化Pipeline和发送流。
4.5 DataNode数据接收服务线程启动（DataXceiverServer & DataXceiver）
当DataNode启动后，其内部的DataXceiverServer组件启动，此线程管理向其所属的DN发送数据的连接建立工作，新连接来时，DataXceiverServer会启动一个DataXceiver线程，此线程负责流向DN的数据接收工作。
4.6 在Pipeline中处理数据的发送和接收
客户端在获取了NameNode分配的文件块的网络位置之后，就可以和存放此块的DataNode交互。

客户端通过SASL加密方式和DN建立连接，并通过pipeline来发送数据。
4.6.1 从pipeline接收数据
pipeline由数据源节点、多个数据目的节点组成，请参考上面的流程图。
位于pipeline中的第一个DataNode会接收到来自客户端的数据流，其内部DataXceiver组件，通过读取操作类型（OP），来区分进行何种操作，如下所示：
  protected final void processOp(Op op) throws IOException {
    switch(op) {
    case READ_BLOCK:
      opReadBlock();
      break;
    //本例中将会使用WRITE_BLOCK指令
    case WRITE_BLOCK:
      opWriteBlock(in);
      break;
    
    //略...
    
    default:
      throw new IOException("Unknown op " + op + " in data stream");
    }
  }
如果OP是WRITE_BLOCK，调用写数据块的方法，此方法会根据数据源是客户端还是其他DataNode、块创建的阶段等条件进行不同的逻辑。
4.6.2 数据在pipeline中流动
在本例中，第一个收到数据的DN会再启动一个blockReceiver线程，以接收实际的块数据，在本地保存了块数据后，其负责向pipeline中的后续DN继续发送块数据。
每次向下游DN节点发送数据，标志着数据目的节点的targets数组都会排除自身，这样，就控制了pipeline的长度。
下游收到块数据的DN会向上游DN或者客户端报告数据接收状态。

这种链式或者序列化的数据转移方式，就像数据在管道中从上游流向下游，所以这种方式称作pipeline。

4.6.3 pipeline的生命周期


在本例中：

DataStreamer线程启动后，pipeline进入PIPELINE_SETUP_CREATE阶段；

数据流初始化后，pipeline进入DATA_STREAMING阶段；

数据发送完毕后，pipeline进入PIPELINE_CLOSE阶段。

客户端在DataStreamer线程启动后，同时启动了一个ResponseProcessor线程，此线程用于接收pipeline中来自下游节点的数据接收状态报告pipelineACK，同时此线程和DataStreamer线程协调管理pipeline状态。
当DataStreamer向pipeline发送数据时，会将发送的数据包（packet）从数据队列（Data Queue）中移除，并加入数据确认队列（Ack Queue）：

//DataStreamer发送数据后，将dataQueue的第一个元素出队，并加入ackQueue
one = dataQueue.getFirst();
dataQueue.removeFirst();
ackQueue.addLast(one);

而当ResponseProcessor收到下游的pipelineAck后，据此确认信息来判断pipeline状态，是否需要重置和重新调整。如果确认信息是下游节点数据接收成功了，就将确认队列（AckQueue）的第一个数据包删除。

//ResponseProcessor收到成功的Ack，就将ackQueue的第一个包移除
lastAckedSeqno = seqno;
ackQueue.removeFirst();
dataQueue.notifyAll();

通过这样的方式，DataStreamer可以确认数据包是否发送成功，也可以确认全部的数据包是否已经发送完毕。

显然，当AckQueue空了，并且已经发送的数据包是块里的最后一个包，数据就发送完毕了。

发送完毕的判断如下所示：
          if (one.lastPacketInBlock) {
            // wait for all data packets have been successfully acked
            synchronized (dataQueue) {
              while (!streamerClosed && !hasError && 
                  ackQueue.size() != 0 && dfsClient.clientRunning) {
                try {
                  // wait for acks to arrive from datanodes
                  dataQueue.wait(1000);
                } catch (InterruptedException  e) {
                  DFSClient.LOG.warn("Caught exception ", e);
                }
              }
            }
            if (streamerClosed || hasError || !dfsClient.clientRunning) {
              continue;
            }
            //在没有错误的情况下，AckQueue为空，并且包one是block的最后一个包，数据就发送完了
            stage = BlockConstructionStage.PIPELINE_CLOSE;
          }
4.7 发送文件操作完成请求（completeFile）
客户端向NameNode发送completeFile请求：

NN收到请求后，验证块的BlockPoolId是否正确，接着对操作权限、文件写锁（write lock）、安全模式、租约、INode是否存在、INode类型等等进行验证，最后记录操作日志并返回给客户端。

4.8 停止文件租约（endFileLease）
客户端在完成文件写操作后，调用leaseRenewer（LR）实例，从LR管理的续约文件表中删除此文件，表明不再更新租约，一段时间后，租约在NN端自然失效。