相对《简单多线程拷贝单文件示例》扩展了任务队列。
主要核心在于将单个大文件分成多份（比如100），形成一个任务，并将任务用链表链接起来，形成一个队列(FIFO)或者栈（无非是顺序不同）。
相对第一版来说，thread_block的定义发生了些变化，但用户接口未变。
typedef struct thread_block
{
int infd; ///<*文件句柄
int outfd;
size_t start_position;///<*文件的写入起始位置
size_t block_size; ///<* 文件写入的终止位置[first ,last)开区间
//struct list_head next;
struct thread_block *next;
}thread_block_t;
定义一个新的结构，原本设想是作为全局变量的。
typedef struct msg_box
{
pthread_mutex_t mutex;
struct thread_block *mblock;
}msg_box_t;
void msgbox_init(msg_box_t *mbox)
{
pthread_mutex_init(&(mbox->mutex),NULL);
mbox->mblock = thread_block_new();
thread_block_init(mbox->mblock,
-1,-1,
0,0);
}
void msgbox_destroy(msg_box_t *mbox)
{
pthread_mutex_destroy(&(mbox->mutex));
free(mbox->mblock);
}
调试信息
void msgbox_printf(msg_box_t *mbox)
{
thread_block_t *block = mbox->mblock;
while(block)
{
printf("start = %d\t end =%d\n",block->start_position,
block->block_size);
block = block->next;
}
}
添加到任务队列和从队列中取任务。
void mpost_task(msg_box_t *mbox,
thread_block_t *msg)
{
pthread_mutex_lock(&(mbox->mutex));
msg->next = mbox->mblock->next;
mbox->mblock->next = msg;
pthread_mutex_unlock(&(mbox->mutex));
}
thread_block_t *mfetch_task(msg_box_t *mbox)
{
thread_block_t *msg = NULL;
pthread_mutex_lock(&(mbox->mutex));
if(mbox->mblock->next)
{
msg = mbox->mblock->next;
mbox->mblock->next = msg->next;
}
pthread_mutex_unlock(&(mbox->mutex));
return msg;
}
此处注意要判断队列是否为空，没有任务返回为NULL，作为线程终止的判断。
分析文件，将文件任务分块，用链表链接起来，v1版本中是分配数组，此处是分配链表节点。同时添加了每个线程处理的大小
size_t block_size = THREADS_BLOCK;
同时需要注意的是，由于一般 对于正数Ｘ有x = a * b +c (a b c >0)，所以分配到块数应该是(a+1).
反映到程序中就是
for(; i <= thread_size;++i)
复制总是少了一些，bug了一个小时才发现。悲催。
void get_thread_task(const char *src,
const char *dst,
msg_box_t *mbox)
{
///打开文件
int infd = open(src,O_RDONLY);
int outfd = open(dst,O_CREAT|O_WRONLY,0644);
if(infd == -1|| -1 ==outfd)
{
printf("error while open file \n");
return;
}
size_t file_size = get_filesize(infd);
size_t block_size = THREADS_BLOCK;
size_t thread_size= file_size / block_size;
printf("filesize = %d\t percent_blocks = %d\n",\
file_size,block_size);
int i = 0;
thread_block_t *block ;
//init-thread-block
for(; i <= thread_size;++i)
{
block = thread_block_new();
thread_block_init(block,
infd,
outfd,
i*block_size,
block_size);
mpost_task(mbox,block);
}
///the last piece
//blocks[i].block_size = file_size%block_size;
block->block_size = file_size%block_size;
}
在v1中，对一个thread_block_t的操作是一个线程函数操作。在v2中，为了程序的流畅性，依然作为一个函数，新的函数名为thread_block_copy，实现与第一版基本一致。
thread_block_copy
int thread_block_copy(thread_block_t *block,char *buf)
{
size_t count = 0;
int ret;
printf("In Thread\t%ld\nstart = %ld\t end = %ld\n",\
pthread_self(),block->start_position,block->block_size);
///lseek到同样的位置
ret = lseek(block->infd,block->start_position,SEEK_SET);
ret = lseek(block->outfd,block->start_position,SEEK_SET);
int bytes_read;
int bytes_write;
while(count < block->block_size)
{
bytes_read = read(block->infd,buf,sizeof(buf));
if(bytes_read >0)
{
printf("thread = %ld\t read = %ld\t count %d\n",\
pthread_self(),bytes_read,count);
count += bytes_read;
//error while read
if((bytes_read == -1)&&(errno !=EINTR))
break;
char *ptr_write = buf;
//悲剧的少了个括号，于是bytes_write == 1
while((bytes_write = write(block->outfd,ptr_write,bytes_read))!=0)
{
if((bytes_write == -1)&&(errno!=EINTR))
break;
if(bytes_write == bytes_read)
break;
else if(bytes_write > 0)
{
ptr_write += bytes_write;
bytes_read -= bytes_write;
}
printf("thread = %ld\t write = %ld\t read %d\n",\
pthread_self(),bytes_write,bytes_read);
}//end-write;
///error while write
if(bytes_write == -1)
break;
}//end-if
}//end-read
return ret;
}
注意：请忽略ret返回值。
再就是线程处理函数了。
/**
* @brief 线程实现函数
* */
void *thread_copy_fn(void *arg)
{
msg_box_t *mbox = (msg_box_t *)arg;
char buf[THREADS_BUFF_SIZE];
int ret;
struct thread_block *block;
while((block = mfetch_task(mbox))!=NULL)
{
ret = thread_block_copy(block,buf);
thread_block_free(block);
sleep(1);
}//end while
printf("#####Thread exit %ld#####\n",pthread_self());
pthread_exit(NULL);
}
由于将大块的实现放到了了thread_block_copy这个函数，所以，显得还是很清新的。
从mbox中取任务，再进行copy处理。
over。
此版需要fixed的地方
1.由于get_thread_task中的打开了文件，所以，必须在拷贝结束之后close，文件fd该存在哪里就成了一个问题，
所以在生成队列的时候，多生成了一个节点，即带头节点的单链表。将fd信息保存到第一个节点当中。所以出现了
很ugly的代码
mbox.mblock->infd = mbox.mblock->next->infd;
mbox.mblock->outfd = mbox.mblock->next->outfd;
这点应该是接口处理得不够好的原因。
2.对于动态的情况来说，不会是parse完成后，生成任务队列之后就不再添加新的任务，链表应该的动态的，即get_thread_task
也应该是个线程，给thread_copy_fn发送消息。此处涉及到任务什么时候真正完成的判断（应该的添加应该通信信号量即可）。
3.对于2的情形，mbox的代码还有改进的空间，因为mbox不涉及到队列（链表）满的判断（参考《消息队列的实现》，用循环数组实现）。