分享一下我写的异步文件I/O的框架（改进版）欢迎大家拍砖

本帖最后由 cookis 于 2010-07-21 00:32 编辑

这是我做CDN系统写的一个独立的框架，不是项目，所以仅做参考，

前一阵翻了一下glibc 中的aio实现，竟然是用 pthread_create 为每一个aio_cb 创建一个dispatched线程，而且有好多条件变量和互斥量，
都是些用户层的调用，感觉不是很高效，所以自己尝试写了一个，经过稳定性测试，效果还不错。

特性
1. 不同于glibc 中 aio的频繁创建销毁线程，而是采用固定的线程池（也可在运行时动态增加）
2. 每个文件句柄有一个优先级，每个句柄关联的任务也有一个优先级，可按优先级进行排序处理
3. 设计原则本着使用者对同一文件访问多次时只打开一次，共享使用一个句柄，这样框架按照句柄归类，线程可按句柄提取任务，不必进行互斥
4. 相同优先级的同一个文件句柄的不同任务按照 I/O的偏移量来排序，减少磁头不必要的移动，更充分地利用vfs的预读功能
5. 采用数组+双向链表+单向链表结构，能够快速定位，快速插入、删除。
6. 内置一个小型的对象池，避免频繁分配小内存。

asynch_file_io.rar (4.1 KB)

下载次数: 8

2010-07-21 00:14

cpp的代码直接贴出来吧

vim set sw=2 ts=2 cinoptions={0,:0,g0,l1,t0,(0

1. #include "ndk/asynch_file_io.h"
   
2. 
   
3. #include <limits.h>
   
4. #include <cassert>
   
5. #include <unistd.h>
   
6. 
   
7. namespace ndk
   
8. {
   
9. void aio_opt_t::reset()
   
10. {
    
11.   this->handle_       = NDK_INVALID_HANDLE;
    
12.   this->opcode_       = -1;
    
13.   this->errcode_      = 0;
    
14.   this->fd_prio_      = 0;
    
15.   this->io_prio_      = 0;
    
16.   this->id_           = -1;
    
17.   this->i_nbytes_     = 0;
    
18.   this->o_nbytes_     = 0;
    
19.   this->offset_       = 0;
    
20.   this->buffer_       = 0;
    
21.   this->aio_handler_  = 0;
    
22.   this->prev_         = 0;
    
23.   this->next_         = 0;
    
24.   this->ptr_          = 0;
    
25.   this->header_       = 0;
    
26. }
    
27. int asynch_file_io::open(size_t thr_num)
    
28. {
    
29.   int size = ndk::max_handles();
    
30.   if (size <= 0) return -1;
    
31. 
    
32.   this->fd_pool_size_ = size;
    
33. 
    
34.   this->fd_pool_ = new aio_opt_t*[this->fd_pool_size_];
    
35.   for (int i = 0; i < this->fd_pool_size_; ++i)
    
36.     this->fd_pool_[i] = 0;
    
37. 
    
38.   return this->activate(ndk::thread::thr_join, thr_num);
    
39. }
    
40. int asynch_file_io::start_aio(ndk::ndk_handle handle,
    
41.                               int *aio_id,
    
42.                               size_t nbytes,
    
43.                               uint64_t offset,
    
44.                               char *buff,
    
45.                               asynch_handler *handler,
    
46.                               int optcode,
    
47.                               int fd_priority,
    
48.                               int io_priority)
    
49. {
    
50.   assert(nbytes > 0);
    
51.   assert(handle != NDK_INVALID_HANDLE);
    
52.   assert(handler != 0);
    
53.   assert(optcode == AIO_READ || optcode == AIO_WRITE);
    
54.   assert(handle >= 0 && handle < this->fd_pool_size_);
    
55.   assert(aio_id != 0);
    
56. 
    
57.   aio_opt_t *aioopt = this->alloc_aio_opt();
    
58.   if (aioopt == NULL) return -1;
    
59. 
    
60.   // guard
    
61.   ndk::guard<ndk::thread_mutex> g(this->queue_list_mtx_);
    
62. 
    
63.   int id = ++this->id_itor_;
    
64.   if (this->id_itor_ == INT_MAX)
    
65.     this->id_itor_ = 0;
    
66. 
    
67.   // assign data struct.
    
68.   aioopt->aio_handler_  = handler;
    
69.   aioopt->handle_       = handle;
    
70.   aioopt->i_nbytes_     = nbytes;
    
71.   aioopt->buffer_       = buff;
    
72.   aioopt->offset_       = offset;
    
73.   aioopt->opcode_       = optcode;
    
74.   aioopt->fd_prio_      = fd_priority;
    
75.   aioopt->io_prio_      = io_priority;
    
76.   aioopt->id_           = id;
    
77.   //
    
78.   *aio_id = id;
    
79. 
    
80.   aio_opt_t *top_node = 0;
    
81.   if (this->fd_pool_[handle] == 0)
    
82.   {
    
83.     // slot is empty.
    
84.     this->fd_pool_[handle] = aioopt;
    
85.     top_node = aioopt;
    
86. 
    
87.     if (this->queue_list_ == 0)
    
88.     {
    
89.       this->queue_list_ = top_node;
    
90.     }else
    
91.     {
    
92.       // keep order.
    
93.       if (top_node->fd_prio_ > this->queue_list_->fd_prio_)
    
94.       {
    
95.         top_node->next_ = this->queue_list_;
    
96.         this->queue_list_->prev_ = top_node;
    
97. 
    
98.         this->queue_list_ = top_node;
    
99.         this->queue_list_->prev_ = 0;
    
100.       }else
     
101.       {
     
102.         aio_opt_t *itor = this->queue_list_;
     
103.         while (itor->next_
     
104.                && (top_node->fd_prio_ <= itor->next_->fd_prio_))
     
105.           itor = itor->next_;
     
106. 
     
107.         top_node->next_ = itor->next_;
     
108.         top_node->prev_ = itor;
     
109.         if (itor->next_)
     
110.           itor->next_->prev_ = top_node;
     
111.         itor->next_ = top_node;
     
112.       }
     
113.     } // end of `if (this->queue_list_ == 0)'
     
114.   }else  // end of `if (this->fd_pool_[handle] == 0)'
     
115.     // link to sub queue
     
116.   {
     
117.     assert(this->queue_list_ != 0);
     
118.     top_node = this->fd_pool_[handle];
     
119.     if (aioopt->io_prio_ > top_node->io_prio_)
     
120.     {
     
121.       // replace old top node.
     
122.       aioopt->fd_prio_ = top_node->fd_prio_; // in order to keep fd sequence.
     
123.       this->fd_pool_[handle] = aioopt;
     
124.       aioopt->header_ = top_node;
     
125. 
     
126.       // reset header.
     
127.       if (this->queue_list_ == top_node)
     
128.         this->queue_list_ = aioopt;
     
129. 
     
130.       // insert new node.
     
131.       aioopt->next_ = top_node->next_;
     
132.       if (top_node->next_)
     
133.         top_node->next_->prev_ = aioopt;
     
134.       aioopt->prev_ = top_node->prev_;
     
135.       if (top_node->prev_)
     
136.         top_node->prev_->next_ = aioopt;
     
137. 
     
138.       // handle sub queue.
     
139.       top_node->next_ = top_node->header_;
     
140.       top_node->header_ = 0;
     
141.     }else  // append to sub queue
     
142.     {
     
143.       if (top_node->header_ == 0)
     
144.       {
     
145.         top_node->header_ = aioopt;
     
146.       }else
     
147.       {
     
148.         if (aioopt->io_prio_ > top_node->header_->io_prio_)
     
149.         {
     
150.           aioopt->next_ = top_node->header_;
     
151.           top_node->header_ = aioopt;
     
152.         }else
     
153.         {
     
154.           aio_opt_t *itor = top_node->header_;
     
155. #if 0
     
156.           while (itor->next_
     
157.                  && aioopt->io_prio_ <= itor->next_->io_prio_)
     
158.             itor = itor->next_;
     
159. #else
     
160.           while (itor->next_)
     
161.           {
     
162.             if (aioopt->io_prio_ > itor->next_->io_prio_)
     
163.               break;
     
164.             if (aioopt->io_prio_ == itor->next_->io_prio_
     
165.                 && aioopt->offset_ < itor->next_->offset_)
     
166.               break;
     
167.             itor = itor->next_;
     
168.           }
     
169. #endif
     
170.           aioopt->next_ = itor->next_;
     
171.           itor->next_ = aioopt;
     
172.         }
     
173.       }
     
174.     }
     
175.   }
     
176. 
     
177.   this->not_empty_cond_.signal(); //
     
178. 
     
179.   return id;
     
180. }
     
181. int asynch_file_io::cancel_aio(ndk::ndk_handle handle, int id)
     
182. {
     
183.   assert(handle != NDK_INVALID_HANDLE
     
184.          && (handle > 0 && handle < this->fd_pool_size_)
     
185.          && id > 0);
     
186. 
     
187.   ndk::guard<ndk::thread_mutex> g(this->queue_list_mtx_);
     
188.   if (this->fd_pool_[handle] == 0)
     
189.   {
     
190.     return this->find_in_running_list(id) == 0
     
191.       ? AIO_CANCELED : AIO_NOTCANCELED;
     
192.   }
     
193. 
     
194.   //
     
195.   aio_opt_t *top_node = this->fd_pool_[handle];
     
196. 
     
197.   if (top_node->id_ == id)
     
198.   {
     
199.     if (top_node->header_ == 0)
     
200.     {
     
201.       if (top_node->prev_)
     
202.         top_node->prev_->next_ = top_node->next_;
     
203.       if (top_node->next_)
     
204.         top_node->next_->prev_ = top_node->prev_;
     
205. 
     
206.       if (this->queue_list_ == top_node)
     
207.         this->queue_list_ = this->queue_list_->next_;
     
208. 
     
209.       this->fd_pool_[handle] = 0;
     
210.     }else // has sub queue
     
211.     {
     
212.       top_node->header_->fd_prio_ = top_node->fd_prio_;
     
213. 
     
214.       this->fd_pool_[handle] = top_node->header_;
     
215.       this->fd_pool_[handle]->header_ = top_node->header_->next_;
     
216. 
     
217.       // insert new node.
     
218.       if (top_node->prev_)
     
219.         top_node->prev_->next_ = this->fd_pool_[handle];
     
220.       this->fd_pool_[handle]->prev_ = top_node->prev_;
     
221.       if (top_node->next_)
     
222.         top_node->next_->prev_ = this->fd_pool_[handle];
     
223.       this->fd_pool_[handle]->next_ = top_node->next_;
     
224. 
     
225.       //
     
226.       if (this->queue_list_ == top_node)
     
227.         this->queue_list_ = top_node->header_;
     
228.     }
     
229. 
     
230.     // release.
     
231.     this->free_aio_opt(top_node);
     
232.   }else if (top_node->header_ != 0)
     
233.   {
     
234.     if (top_node->header_->id_ == id)
     
235.     {
     
236.       aio_opt_t *item = top_node->header_;
     
237.       top_node->header_ = item->next_;
     
238.       this->free_aio_opt(item);
     
239.     }else
     
240.     {
     
241.       aio_opt_t *itor = top_node->header_;
     
242.       for (; itor->next_ != 0; itor = itor->next_)
     
243.       {
     
244.         if (itor->next_->id_ == id)
     
245.         {
     
246.           aio_opt_t *item = itor->next_;
     
247.           itor->next_ = item->next_;
     
248.           this->free_aio_opt(item);
     
249.           break;
     
250.         }
     
251.       }
     
252.     }
     
253.   }
     
254. 
     
255.   return this->find_in_running_list(id) == 0 ?
     
256.     AIO_CANCELED : AIO_NOTCANCELED;
     
257. }
     
258. int asynch_file_io::svc()
     
259. {
     
260.   while (1)
     
261.   {
     
262.     aio_opt_t *running_list = 0;
     
263.     {
     
264.       ndk::guard<ndk::thread_mutex> g(this->queue_list_mtx_);
     
265.       while (this->queue_list_ == 0)
     
266.         this->not_empty_cond_.wait(0);
     
267. 
     
268.       running_list = this->pop_some_request(running_list, 5);
     
269. 
     
270.       this->enqueue_to_running_list(running_list);
     
271.     }
     
272.     this->handle_aio_requests(running_list);
     
273.   }
     
274.   return 0;
     
275. }
     
276. aio_opt_t *asynch_file_io::pop_some_request(aio_opt_t *pop_list, int num)
     
277. {
     
278.   /**
     
279.    * Pop out from the front of the queue some requests.
     
280.    * If the slot of <fd> has a sub-queue, then pop all of this sub-queue.
     
281.    */
     
282.   aio_opt_t *pop_list_itor = 0;
     
283.   aio_opt_t *top_node = this->queue_list_;
     
284. 
     
285.   for (int i = 0; i < num && top_node != 0; ++i)
     
286.   {
     
287.     this->fd_pool_[top_node->handle_] = 0;
     
288.     // link top node.
     
289.     if (pop_list == 0)
     
290.     {
     
291.       pop_list = top_node;
     
292.       pop_list_itor = pop_list;
     
293.     }else
     
294.     {
     
295.       pop_list_itor->next_ = top_node;
     
296.       pop_list_itor = pop_list_itor->next_;
     
297.     }
     
298. 
     
299.     // link sub queue.
     
300.     aio_opt_t *node = top_node->header_;
     
301. 
     
302.     /**
     
303.      * Important !!!.
     
304.      * because pop_list_itor point to top node and
     
305.      * pop_list_itor->next_ will be repoint, so we must
     
306.      * store next top node.
     
307.      */
     
308.     top_node = top_node->next_;
     
309. 
     
310.     for (; node != 0; ++i, node = node->next_)
     
311.     {
     
312.       pop_list_itor->next_ = node;
     
313.       pop_list_itor = pop_list_itor->next_;
     
314.     }
     
315.   }
     
316.   this->queue_list_ = top_node;
     
317.   if (this->queue_list_ != 0)
     
318.     this->queue_list_->prev_ = 0;
     
319.   if (pop_list_itor)
     
320.     pop_list_itor->next_ = 0;
     
321. 
     
322.   return pop_list;
     
323. }
     
324. void asynch_file_io::handle_aio_requests(aio_opt_t *running_list)
     
325. {
     
326.   for (aio_opt_t *aioopt = running_list; aioopt != 0; aioopt = aioopt->next_)
     
327.   {
     
328.     int result = 0;
     
329.     void (asynch_handler::*callback)(const aio_opt_t *) = 0;
     
330.     switch(aioopt->opcode_)
     
331.     {
     
332.     case AIO_READ:
     
333.       do
     
334.       {
     
335.         result = ::pread(aioopt->handle_,
     
336.                          (void*)aioopt->buffer_,
     
337.                          aioopt->i_nbytes_,
     
338.                          aioopt->offset_);
     
339.       }while (result == -1 && errno == EINTR);
     
340.       callback = &asynch_handler::handle_aio_read;
     
341.       break;
     
342.     case AIO_WRITE:
     
343.       do
     
344.       {
     
345.         result = ::pwrite(aioopt->handle_,
     
346.                           (const void*)aioopt->buffer_,
     
347.                           aioopt->i_nbytes_,
     
348.                           aioopt->offset_);
     
349.       }while (result == -1 && errno == EINTR);
     
350.       callback = &asynch_handler::handle_aio_write;
     
351.       break;
     
352.     default:
     
353.       assert(0);
     
354.     }
     
355.     if (result >= 0)
     
356.     {
     
357.       aioopt->errcode_  = 0;
     
358.       aioopt->o_nbytes_ = result;
     
359.     }else
     
360.     {
     
361.       aioopt->errcode_  = errno;
     
362.     }
     
363.     (aioopt->aio_handler_->*callback)(aioopt);
     
364.     this->dequeue_from_running_list(aioopt);
     
365.   }
     
366.   this->free_aio_opt_n(running_list);
     
367. }
     
368. void asynch_file_io::enqueue_to_running_list(aio_opt_t *running_list)
     
369. {
     
370.   assert(running_list != 0);
     
371.   ndk::guard<ndk::thread_mutex> g(this->running_list_mtx_);
     
372. 
     
373.   aio_opt_t *aioopt = 0;
     
374.   if (this->running_list_ == 0)
     
375.   {
     
376.     aioopt = this->alloc_aio_opt_i(this->free_running_list_);
     
377.     aioopt->ptr_ = running_list;
     
378.     this->running_list_tail_ = aioopt;
     
379.     this->running_list_ = aioopt;
     
380.     running_list = running_list->next_;
     
381.   }
     
382. 
     
383.   aio_opt_t *itor = running_list;
     
384.   for (; itor != 0; itor = itor->next_)
     
385.   {
     
386.     aioopt = this->alloc_aio_opt_i(this->free_running_list_);
     
387.     aioopt->ptr_ = itor;
     
388.     this->running_list_tail_->next_ = aioopt;
     
389.     this->running_list_tail_ = aioopt;
     
390.   }
     
391. }
     
392. void asynch_file_io::dequeue_from_running_list(aio_opt_t *aioopt)
     
393. {
     
394.   ndk::guard<ndk::thread_mutex> g(this->running_list_mtx_);
     
395.   assert(this->running_list_ != 0);
     
396. 
     
397.   aio_opt_t *itor = this->running_list_;
     
398.   aio_opt_t *free_ptr = 0;
     
399.   if (itor->ptr_ == aioopt)
     
400.   {
     
401.     free_ptr = itor;
     
402.     this->running_list_ = itor->next_;
     
403.   }else
     
404.   {
     
405.     for (; itor->next_ != 0; itor = itor->next_)
     
406.     {
     
407.       if (itor->next_->ptr_ == aioopt)
     
408.       {
     
409.         if (itor->next_ == this->running_list_tail_)
     
410.           this->running_list_tail_ = itor;
     
411.         free_ptr = itor->next_;
     
412.         itor->next_ = itor->next_->next_;
     
413.         break;
     
414.       }
     
415.     }
     
416.   }
     
417.   assert(free_ptr != 0);
     
418.   free_ptr->next_ = 0;
     
419.   this->free_aio_opt_i(free_ptr, this->free_running_list_);
     
420. 
     
421.   if (this->running_list_ == 0
     
422.       || this->running_list_->next_ == 0)
     
423.     this->running_list_tail_ = this->running_list_;
     
424. }
     
425. int asynch_file_io::find_in_running_list(int id)
     
426. {
     
427.   ndk::guard<ndk::thread_mutex> g(this->running_list_mtx_);
     
428.   if (this->running_list_ == 0)
     
429.     return 0;
     
430. 
     
431.   aio_opt_t *itor = this->running_list_;
     
432.   for (; itor != 0; itor = itor->next_)
     
433.     if (itor->ptr_->id_ == id) return -1;
     
434.   return 0;
     
435. }
     
436. void asynch_file_io::free_aio_opt_n(aio_opt_t *p)
     
437. {
     
438.   ndk::guard<ndk::thread_mutex> g(this->free_list_mtx_);
     
439.   aio_opt_t *tail = p;
     
440.   --this->queue_list_size_;
     
441.   while (tail->next_)
     
442.   {
     
443.     tail = tail->next_;
     
444.     --this->queue_list_size_;
     
445.   }
     
446.   tail->next_ = this->free_list_;
     
447.   this->free_list_ = p;
     
448. }
     
449. #ifdef NDK_DUMP
     
450. void asynch_file_io::dump()
     
451. {
     
452. }
     
453. #endif
     
454. } // namespace ndk

复制代码

作者: cookis   发布时间: 2010-07-21

作者: rain_fish   发布时间: 2010-07-21

作者: cookis   发布时间: 2010-07-21

作者: rain_fish   发布时间: 2010-07-21

作者: cookis   发布时间: 2010-07-21

QUOTE:

回复  rain_fish


    好吧，看这里
cookis 发表于 2010-07-21 10:39

    呵呵，

作者: rain_fish   发布时间: 2010-07-21

作者: rain_fish   发布时间: 2010-07-21

作者: peidright   发布时间: 2010-07-21