一 前言
看了一下，距离上一篇博客的发表已经过去了4个月，时间过得好快啊。本篇博客是JDFS系列的第三篇博客，JDFS的目的是为了实现一个分布式的文件管理系统，前两篇实现了基本的上传、下载功能，但是那还不能算作分布式文件管理。本篇博客将在前两篇的基础上增加一系列分布式的功能，比如流式云存储，就是客户端把本地的文件切分成若干片后，以冗余的方式存储到分布式数据结点上；所谓的流式传递读者可以在网上搜索一下HDFS的流式传递，基本上就是那个意思，正文中会详细介绍这个，此处不再赘述。除了分布式存储外，当然我们还得支持客户端显示特定文件的信息，比如这个文件有多少个block组成，每一个block存储在哪几个数据结点上等，我们不妨把文件的这些信息称之为文件元信息，元信息是存储在name node结点上面的。然后有了云存储、文件元信息的显示，当然少不了文件的下载，文件的下载会从不同的数据结点下载文件的block，然后再本地把所有的block合并成一个完整的文件。最后支持删除操作，客户端向name node结点发起删除文件的操作，name node会通知所有存储该文件的数据结点删除对应的block,最后name node结点删除存储对应文件的元信息。
上一段提到了三个角色：客户端、数据结点(data node)、name node,。所有的操作都有客户端发起，文件的正文信息冗余地存储在data node上，而文件的元信息存储于name node上。说到这，读者朋友们可能会感觉很熟悉，没错，HDFS在存储文件时也是这么干的。本篇博客将会聚焦于文件的“流式云存储”，而其他功能将会在本系列博客的第四篇进行介绍。JDFS系列博客截止目前一共三篇，内容上相互承接，因此如果读者朋友是第一次阅读本篇博客的话，笔者建议大家先阅读前两篇博客，以保持连贯性，链接在下面：
JDFS相关代码已经更新到github, 其链接地址请点击我, 如果读者感觉本文对你有所帮助，不妨给JDFS的github点个star, 嘿嘿~~
Note: 网络问题是一个非常复杂的问题，再加上多线程，epoll调试起来非常困难。截止目前JDFS的功能并不是十分完美，偶尔会出错，但是这些错误很难重现，一时半会并不能全部解决，但是这并不影响对JDFS总体设计的介绍，总体的框架是没问题的。这些潜在的问题会在后续版本想办法解决，争取早日得到一个稳定版的JDFS.
PS: 本篇博客是博客园用户“cs小学生”的原创作品，转载请注明原作者和原文链接，谢谢。
二 实验平台
前文提到了分布式结点，包括data node， name node等，目前笔者并没有条件搭建一个真正的分布式环境。但是好在有免费版的vmware player, 所以笔者用vmware 创建了4个虚拟机，每一个虚拟机模拟一个结点，四个虚拟结点的名字分别是data node 1, data node 2, data node3, data node4， 为了方便后续代码的处理，四个虚拟结点分别分配了一个串号：1,2,3,4 。串号唯一标志了一个虚拟结点。为什么不用ip地址来标记呢？因为结点的ip 地址并不是一直保持不变的。根据笔者这段时间做实验的经验来看，每隔几天，虚拟机的ip地址都会变好一次，使用ifconfig命令就可以看到虚拟ubuntu的ip v4地址。
在笔者的实验里：data node 2上面跑的是name node server,充当name node角色， data node 1,， 3，4 上面跑的是data node server代表了数据结点， 其中data node 3上面同时还运行着客户端结点。笔者只用了四个结点来测试，其原因是JDFS代码刚刚更新，先从简单入手，在较少的结点上进行测试，方便些，等功能稳定了再在多个结点上进行测试。
实验平台截图如下：
如上图所示，是data node server， name node server都启动后的截图，上面三个是数据结点， 最下面的是name node。
三 一些遗留的问题
笔者在上篇博客的基础上添加分布式功能的时候，发现了几个bug, 在介绍正式功能前我们先来看看这几个bug
1. 都是水平触发惹的祸
熟悉linux网络编程的人大都知道，epoll是高并发服务端设计的首选，epoll中有两个概念：水平触发、边缘触发。在epoll中我们通过不断地调用epoll_wait()函数来检测当前监听的<ip:port>上面是否有读写事件发生，水平触发和边缘触发的区别就在于调用epoll_wait()的时候如何报告读写事件。以报告读事件为例（客户端发数据给服务端），对于水平触发：如果当前的socket fd内部的缓冲区中有数据，epoll_wait()就会无条件地告诉调用者：hi, 缓冲区里现在有数据，快来读啊。即使服务端已经指派了一个线程来处理这个socket fd上的数据，只要epoll_wait()执行时，数据还没被处理完，epoll_wait()就会报告读事件。而对于边缘触发：在第一次报告socket_fd上有读数据后，以后epoll_wait()执行时，即使socket fd内部的缓冲区里有数据也不报告，只有缓冲区再一次变空后，再有数据到来才会报告读事件；简言之，只有缓冲区的状态从“空”变为“不空”才会报告事件，所以如果线程读取一个缓冲区的数据，如果没有读完，那么缓冲区让然处于“不空”的状态，因此没有发生“状态”的改变，所以不会报告事件。
我们假想一种场景，在水平触发的情况下，epoll_wait()报告了一个读事件，并且把该事件派发给线程1来处理，然后epoll_wait()再一次执行的时候线程1没有把数据处理完，于是epoll_wait()再次报告读事件，并且派发给线程2来执行。加入这一片数据逻辑上只应该由同一个线程来处理，那么此时错误就会发生。笔者写完流式云存储部分的功能测试的时候，一开始老是出错，发现线程读的数据始终少了一部分，而且少的这部分数据恰好就是http_request头部的长度（注：头部是笔者自定义的一个数据结构，客户端每次发送的数据都会带一个头部，用来描述数据正文的相关信息，比如正文长度等，同样服务端的线程在读socket fd的数据时，也是首先读取头部，解析后再读取正文）。
说道这，估计读者已经猜出了原因了，笔者之前的代码使用的是水平触发，假设线程1读取头部解析到需要读取4096字节的正文数据后就开始使用recv()来读取相应的数据，与此同时，epoll_wait()检测到同一个socket fd上还有数据，于是把该读事件作为作业加入到作业队列里，假设线程2竞争到该作业，于是开始读取一个头部（实际上是正文数据），解析后发现不是有效的头部于是丢弃该作业，而线程1由于线程2已经读取了一个头部，而少读取了一个头部，这就是前文所述bug的来龙去脉。
那么读者可能要问了之前怎么没有发现这个bug呢？答案是之前不存在发生这个bug的机会，在本篇博客之前，在做测试时，只有一个服务端和客户端，这样的话，很可能下一次epoll_wait()被调用的时候线程1已经把数据读取完毕了，问题主要发送在流式传递(详情见后文)的过程中：客户端--->data node 1--->data node3--->data node 4,笔者实验的时候问题主要发生在数据流式传递到data node 3和 data node 4的情况，应该是随着结点数目的增多以及代码逻辑的复杂性的增加导致了线程还没把数据读取完毕另外一个线程就也开始读取同一个socket fd上的数据了。
于是笔者后来采用了边缘触发的方式，使用边缘触发后，上述的bug大大减少了，但是在很少的情况下让然会出现数据读取不全的问题。经过反复测试与思考，笔者想出了一个可能导致出错的场景(后来验证确实是这个原因)：我们假设有一片数据[a,b] a表示起始位置，b表示结束位置。由于网络的原因，这片数据不是一下就到达，而是分为两次[a,m]和[m+1,b]到达，假设在[a,m]到达后，在线程1读取完[a,m]后[m+1,b]由于网络延迟或者其他原因没有到达，这时候socket fd的内部缓冲区已经空了，过了一会[m+1,b]已经到达了，结果缓冲区的状态由“空”变为“不空”，这种情况下即使是边缘触发也会报告读事件，此时线程1卡住了，等待数据[m+1,b]而线程池里空闲的线程检测到刚才epoll报告的读事件也开始读取同一个socket fd的数据，于是bug又发生了，虽然这次发生的概率比较低。
针对这个bug场景，解决的办法是使用EPOLLONESHOT标志，使用了这个标志后，不管缓冲区状态是否发生变化，都不会报告事件，这样线程1可以安心地把数据读完，读完后再把EPOLLONESHOT标志复位，这样epoll_wait()又可以愉快地报告读事件了。
2. malloc & fopen: 都是越界写数据惹的祸
这个bug的截图当时笔者恰好保存了，我们先来看下截图吧：
从图上看是malloc相关的错误，很奇怪的是经过笔者的反复调试，最终定位的发生错误的位置在一个fopen语句，malloc的报错为什么会在fopen的时候出现呢。网上搜索了一下好像fopen内部会调用malloc()函数，但是笔者反复检查了fopen的参数，也没有传递非法参数啊。后来看了下fopen的前边有一个数据写入的操作，写入的是一个用malloc分配的内存，然后看了下malloc分配该数据的地方，发现分配的长度不对，长度应该是 xx*sizeof(int)错写成了xx,改了后，于是错误就消失了。猜想原因可能是这样的：写数据的地方越界了，而fopen内部再一次调用malloc时发现了某些字段被越界写破坏了，于是就报错了。
3. ftell: 都是磁盘缓冲惹的祸
在添加分布式功能的时候，有一个操作是：在本地把文件分成若干个block(假设5个block),并且存储于磁盘，然后客户端在挨个读取每个block,用ftell获取文件长度，但是发现在处理block1-4都正常，而ftell block5的长度总是少了一部分。于是笔者另外单独写了一个main函数来ftell block5的长度是正确的，只有在JDFS里ftell这个block长度才会出错，经过一番思考，灵光乍现，可能是内存里有一部分在缓冲里没有写到磁盘里，JDFS是分片后立即ftell block的长度，还没来得及写入磁盘，而单独写的main函数是执行时已经过了一段时间，缓冲已写入磁盘。然后再检测分片函数时发现，原始文件分片存储后没有调用fclose()函数，加上fclose()后bug就消失了，因为fclose会刷新缓冲区的数据将之存储到磁盘里面。
4. 线程池：都是变量初始化惹的祸
还有一个是线程池相关的bug，这里简单的提一下，原代码里面是在创建了线程之后，才开始初始化诸如mutex变量，这样是不对的，应该先初始化后再创建线程，否则读到为初始化的数据导致逻辑错误。
四 流式云存储
哎呀，总算到了讲正题的地方了，哈哈~~。
http_request_buffer结构体是JDFS最重要的一个数据结构，其中的一个字段request_kind代表了不同的请求类型，客户端向服务端发请求，服务端向另外一个服务端发请求都得带上一个头部，最初支持查询、上传、下载三个功能，这次更新后添加了很多其他的功能，让我们先来看看request_kind现在都有哪些功能吧。下面是截取的该数据结构的代码注释：
/*
** request_kind, 0:query file size 1: upload 2: download 3,4: ack
** 5: query node list alive 6: query one certain node alive
** 7: ack alive of one certain node
** 8: query ip form node serial 9: return to client,not found
** 10: stream transform
** 11: client post clousd store,giving the meta info 12: failed ack
** 13: query file meta info, read or delete 14: not exsists ack
** 15: update meta info 16: update failed ack
** 17: wait meta complete 18: not complete
** 19: delet meta file
** 20: delete file
*/
其中主要是0,1,2,5,6,8,10,11,13,15,17,19,20. 本文涉及到的主要是：5,6,8,10,11,15,17。下面我们就开始分别详细的介绍~~
1. 流式云存储总体流程的简介
我们先来描述一下从客户端发出请求到服务端接收完数据，这一个过程中都发生了哪些事情。下面我们简称client 为C, data node 1,2,3为D1, D2, D3, name node server为 NS. 客户端要处理的文件名为CRLS.pdf(其实就是算法导论，哈哈)
我们的目标是：C把CRLS.pdf切分成m份，然后向NS查询当前虚拟集群里还“活着”的data node list，然后为每一份数据指定存储于哪几个数据结点(例如CRLS.pdf.blk1存储于D1, D2, D3), 以CRLS.pdf存储于D1, D2, D3为例，C把CRLS.pdf.blk1按照10K的大小切分成n个分片，把每一分片上传给D1, 以分片S1为例：前文我们提到过正文之前有一个头部http_request_buffer,我们在这个头部里记录结点D2,D3的信息，于是D1从C接收到分片S1后，紧接着检测到还需要“流式”传递给D2,D3,于是D1把该片数据传递给D2,当然传递前清除头部的D2,以免后续重复传递给D2,同理，D2接收到数据后在传递给D3.经过一段时间后CRLS.pdf.blk1传递完毕，CRLS.pdf.blk2,3,...m按照同样的流程处理，最终整个CRLS.pdf被分成了m份，冗余地存储于不同的数据结点上。而对于CRLS.pdf被分成了多少个block,每一个block存储于哪些数据结点这样的元信息存储于NS上，这其实就是模仿了HDFS的处理过程，在这里我们向HDFS致敬。PS: 笔者写JDFS的目的不是为了实现一个和HDFS功能类似的系统，而是锻炼运用若干技术实现解决一个实际问题的能力，或者说是锻炼设计软件架构从而实现一个完整的系统而不是简单写一个具有单一功能的程序的能力。
每一个data node在接收完某个block比如CRLS.pdf.blk1的时候需要向NS报告“我已经接收完毕CRLS.pdf的第一个block了，请将之记录到对应文件的元信息里吧”。NS端记录着CRLS.pdf的每一个block应当存储于哪些data node里，当所有的data node都接收完毕对应的block后，NS就会在该文件的元信息里标记为已完成存储。这样后续C就可以请求该文件的元信息，并根据元信息到不同的data node下载对应的block,然后合并block还原出原来的文件，后续还可以请求NS从虚拟集群里删除这个文件。
那么对于特定的data node,比如D1接收完CRLS.pdf.blk1,怎么才算接收完毕呢？对于data node 1，线程池里同时有N个线程在处理接收CRLS.pdf.blk1的不同的分片。笔者是这样的处理的：在发送数据的头部，记录该分片是所在block的第几个分片，以及所在的block是原始文件的第几个block. 当D1检测到所接收的数据分片是对应的block的最后一片分片的时候，D1就会等待其他该block的其他分片被处理完毕，然后发消息给NS请求更新文件的元信息。为了支持该操作，我们定义了一个二维数组node_block_meta_info[m][n], 其中m是原始文件被分成的block的数目，n是特定block对应的分片的信息。如果某个线程接收到第一个block的第一个分片，那么它就负责分配node_block_meta的第一维空间，如果线程接收到了第x个block的第一个分片，那么负责创造对应的第二维存储空间，也即node_block_meta_info[x]=....malloc()....
在上文的基础上，假设某个线程接收到CRLS.pdf.blk1的第m片数据，于是就设置node_block_meta_info[1][m]=1,意为该分片被成功接收了，当然必须是node_block_meta_info!=NULL&&node_block_meta_info[1]!=NULL的时候才能写入，否则应该等待相应的内存被分配后在写入。这样一来，线程检测某个block是否被全部接收就容易了，直接遍历node_block_meta_info数组就行了。关于node_block_meta_info的free，也很容易，接收完对应的block比如x，free x位置处的内存，整个文件都接收完毕，就free第一维的node_block_meta_info。
关于谁来创建文件的问题，之前由于客户端是确认一个分片到达后再上传下一个分片，因此服务端只要简单判断磁盘里是否有该文件，没有则创建就可以了。加入分布式功能后，D1往D2或者D3流式传递数据就不一定是串行的，有可能第1个分片还没到达，第2个分片已经接收完毕了，那么文件由谁来创建的？类似地，我们也让接收第一个分片的线程创建，其他的线程等待，这样会不会经常卡住其他线程呢？基本不会，比如文件被分为5个block，则顶多卡住其他线程5次，而一个block被分成上百个分片，这样一来卡的情况可以忽略不计。
接下来我们分别从客户端、name node、data node的角度来分析如何实现流式云存储
2 客户端逻辑
客户端实现了如下几个函数，具体代码详见前言里的github链接
int JDFS_cloud_query_node_list( char *server_ip, int server_port, node_list *nli);
int JDFS_cloud_store_file(char *file_name, int flag);
int JDFS_cloud_store_one_block(char *file_name, int block_num,char nodes_8_bits, node_list *nli, int total_num_of_blocks);
int Split_file_into_several_parts(char *file_name, int part_size);
int Extract_part_file_and_store(FILE *fp,char *new_file_name,int range_begin,int range_end);
int JDFS_wait_file_transform_completed(char *file_name, char *ip_str, int port);
客户端的逻辑比较简单，JDFS_cloud_store_file()是客户端执行的最初入口，参数flag如果为0表示文件不用切分为若干block，否则按照实现定义的规则切分文件为若干个block。该函数会首先调用JDFS_cloud_query_node_list()函数，该函数会向name node server查询当前集群里活跃的data node列表，该列表存储在类型为node_list类型的参数里。这个时候客户端得到了活跃的结点列表，也知道了应该把文件分为几个block,接下来就要为每个block选取若干个结点来存储了，截止目前笔者简单地为每个block选取前3个data node,后续稳定后，会实现一个函数按照特定的规则为每一个block选取相应的结点。查询活跃结点的截图如下：
紧接着，调用Split_file_into_several_parts()函数和Extract_part_file_and_store()函数将原始文件进行切分。然后调用JDFS_cloud_store_one_block()把某个特定的block上传给选择的data node,注意参数nodes_8_bits是一个字符型数据，如果某个bit被设置了就意味着对应的data node 也需要接收此block,服务端的流式传递需要查询这个数据。最后客户端传完数据后，data node之间的流式传递可能还没有结束，此时客户端调用JDFS_wait_file_transform_completed()函数等待文件全部传递完毕，该函数内部会每隔一段时间向name node请求标记该文件为已完成，name node会读取文件元信息，如果确实完成了全部的存储，就会标记为完成，并告诉客户端我已经更新状态为完成了，否则不做更改。
下图是data node接收到数据后的截图：
下面是传输过程中命令行打印的一些消息的截图：
3. Name node 端的逻辑
name node端实现了以下几个函数
int make_socketfd_nonblocking(int socket_fd);
void *Http_server_callback_query_nodelist(void *arg);
int DataNode_alive_or_not(char *ip, int port);
void *Http_server_callback_query_ip_from_node_serial(void *arg);
void *Http_server_callback__cloudstr_meta(void *arg);
void *Http_server_callback_update_meta_info(void *arg);
void *Http_server_callback_wait_meta_complete(void *arg);
由于改版后服务端使用了边缘触发，边缘触发下数据读取最好使用非阻塞方式，因此make_socketfd_nonblocking()就是用来设置对应的socket fd为非阻塞模式的。Http_server_callback_query_nodelist()用来响应服务端查询活跃的data node结点列表，name node会首先读取配置文件里记录的所有存在的data node的ip,port,然后它会调用DataNode_alive_or_not()函数检测ip,port对应的data node是否活跃，搜集完所有的活跃结点后，发送给客户端。前文中提到，某一个data node接收完数据后需要把数据流式传递给其他data node, 所以需要先向name node查询对应串号的data node的ip,port, 而 Http_server_callback_query_ip_from_node_serial()就是用来响应这个请求的。Http_server_callback__cloudstr_meta()函数用来响应客户端请求存储文件的请求用的，具体是用来创建该文件对应的元信息文件。前文说过，某个data node存储完一个block后需要告诉name node更新对应的元数据信息，而Http_server_callback_update_meta_info()就是用来响应该请求的。前文还说过，客户端发送完所有block后，需要间隔式地发送请求给name node请求标记文件为已完成存储，而Http_server_callback_wait_meta_complete()就是用来响应这个请求的。
4. data node端的逻辑
对于data node端，主要更新了void *Http_server_callback_upload(void *arg);函数以支持流式传递，并且增加了Http_stream_transform_file(callback_arg_upload *cb_arg_upload, char *namenode_ip, int namenode_port)来做具体的流式传递工作，所谓流式传递即为某个data node接收完一片数据后，紧接着传递给另一个需要该片数据的data node.
4.1 void *Http_server_callback_upload(void *arg)
该函数用来接收客户端或者其他data node发送过来的数据分片，为支持流式传递，做了一些更改。
首先前文讨论过关于由哪个线程创建文件的问题以及如何确定某个block的分片全部接收完毕，这部分都是在data node server端来做的，首先该函数如果检测到当前接收到的分片是某个block的第一个分片，那么就创建文件，并且为node_block_meta_info数组分配内存空间，并做相应的初始化。接收完数据后，该函数会调用Http_stream_transform_file()把该片数据传递给下一个需要该数据的data node, 传递完成后，需做如下判断：
如果该片数据是对应block的最后一片数据，那么就等待该block的所有数据片接收完毕，然后释放node_block_meta_info数组对应的内存空间，如果该片数据是最后一个block的最后一个分片，那么就等待所有block传递完毕后彻底释放node_block_meta_info数组。
4.2 int Http_stream_transform_file(callback_arg_upload *cb_arg_upload, char *namenode_ip, int namenode_port);
该函数的功能也很简明，首先检测是否存在下一需要改数据分片的data node,如果有的话，根据该data node的串号，向name node查询对应的ip地址和端口号port,然后把该数据分片发送给<ip,port>标志的data node.
五. 结束语
至此本篇博客就结束了，主要是在上一篇的基础上增加了分布式存储的功能，本文简要的介绍了流式云存储的执行流程，以及三个角色：name node, data node, client如何共同协作完成这一功能的。具体的代码逻辑请见前言里给出的github链接。当然目前JDFS的功能并不是完美稳定的，主要是因为网络+线程池+epoll使得一些潜在的bug很难重现与调试；当然后续会不断地对JDFS进行优化，调试，争取早日达到稳定状态。
联系方式：联系方式：https://github.com/junhuster/