多线程写文件问题，求指点

colrain_h

本人是Linux C初学者，现在需要写一个实时读取加速网卡数据并将其负载写盘的程序。我现在写的程序实现4个线程同时从网卡接收数据流，并为每一个流建立一个文件写入其负载。目前这种模式已经跑通。但是随着数据量加大，这种模式一般写到1000个文件左右，文件指针就会耗尽，程序报fopen failed! 请各位高手给想想办法，改变一下程序结构，让这个程序既能实时按每个流一个文件的形式落下负载，同时又不把文件指针耗尽。代码如下：

1. #include <stdio.h>
   
2. #include <stdlib.h>
   
3. #include <string.h>
   
4. #include <assert.h>
   
5. #include <time.h>
   
6. #include <fcntl.h>
   
7. #include <sys/types.h>
   
8. #include <unistd.h>
   
9. #include <sys/stat.h>
   
10. #include <sys/time.h>
    
11. #include <arpa/inet.h>
    
12. #include <pthread.h>
    
13. #include "libpag.h"
    
14. #include "eb_pcap.h"
    
15. 
    
16. typedef struct receive_thread_info {
    
17.     pthread_t tid;
    
18.     int sid;
    
19.     unsigned int file_start_num;
    
20.     unsigned int file_finish_num;
    
21. } receive_thread_info_t;
    
22. 
    
23. /*
    
24. * create a file
    
25. */
    
26. static FILE* get_file(receive_thread_info_t *info_p)
    
27. {
    
28.     char file_name[16];
    
29.     FILE* fp;
    
30.     memset(file_name, 0, 16);
    
31.     sprintf(file_name, "%d_%d", info_p->sid, info_p->file_start_num);
    
32.     strcat(file_name, ".txt");
    
33.         //为每个流建立一个文件
    
34.     fp = fopen(file_name, "wb");
    
35.     if (fp == NULL) {
    
36.         printf("%s open faild!\n", file_name);
    
37.         exit(1);       
    
38.     }
    
39.     return fp;
    
40. }
    
41. 
    
42. /*
    
43. * write packet to the file
    
44. */
    
45. static void write_pkt(struct pktinfo *pktinfo_ptr)
    
46. {
    
47.     struct tcp_stream *tcp = pktinfo_ptr->pdata;
    
48.     FILE *fp = (FILE *)tcp->appinfo;
    
49.     struct half_stream *half_p = pag_gethalfstream(pktinfo_ptr);
    
50.     if (half_p == NULL) {
    
51.         return;
    
52.     }
    
53.    //写入负载
    
54.     if (half_p->count_new > 0) {
    
55.         fwrite(half_p->data, half_p->count_new, 1 , fp);
    
56.         fflush(fp);
    
57.     }
    
58. }
    
59. 
    
60. static void deal_stream(struct pktinfo *pktinfo_ptr, receive_thread_info_t *info_p)
    
61. {
    
62.     struct tcp_stream *tcp;
    
63.     FILE *fp;
    
64.    
    
65.     tcp = pktinfo_ptr->pdata;
    
66.     if (tcp->appinfo != NULL) {
    
67.         write_pkt(pktinfo_ptr);            
    
68.     } else {
    
69.             fp = get_file(info_p);
    
70.             info_p->file_start_num++;
    
71.             tcp->appinfo = (void *)fp;
    
72.             write_pkt(pktinfo_ptr);
    
73.     }
    
74.    
    
75.     if (pag_isstreamclosed(pktinfo_ptr)) {
    
76.         if (tcp->appinfo != NULL) {
    
77.             fp = (FILE *)tcp->appinfo;
    
78.             fclose(fp);
    
79.             info_p->file_finish_num++;
    
80.             tcp->appinfo = NULL;
    
81.         }
    
82.         pag_delstream(pktinfo_ptr->pdata);
    
83.     }
    
84. }
    
85. 
    
86. /*
    
87. *receive thread, get stream infomation
    
88. */
    
89. void * receive_pthread(void *arg)
    
90. {
    
91.     receive_thread_info_t *info_p = (receive_thread_info_t *)arg;
    
92.     struct pktinfo *pktinfo_p = NULL;
    
93. 
    
94.    
    
95.     while (1) {
    
96.             /* 从网卡的一个数据流队列中读取数据流信息，可能是流也可能是包信息 */
    
97.             pktinfo_p = pag_getstream(info_p->sid);
    
98.              /* 判断：如果是udp 包则不处理 */
    
99.             if (pktinfo_p == NULL || pktinfo_p->ipproto != 0x06) {
    
100.                 usleep(10);
     
101.                 continue;
     
102.             }
     
103.             /* 对每个tcp包进行处理 */
     
104.             deal_stream(pktinfo_p, info_p);
     
105.     }
     
106.     return NULL;
     
107. }
     
108. 
     
109. int main(int argc, char **argv)
     
110. {
     
111.     int i, ret;
     
112.     void *value_ptr;
     
113.     int sid_counter = 4;
     
114.     if ((ret = pag_open()) == -1){
     
115.             printf("pag_open error\n");
     
116.             exit(1);
     
117.     }
     
118.     pag_setlinknum(500);
     
119.     receive_thread_info_t * info_p = (receive_thread_info_t *)calloc(sid_counter,
     
120.             sizeof(receive_thread_info_t));
     
121.     if (info_p == NULL) {
     
122.         printf("calloc error\n");
     
123.         exit(1);
     
124.     }
     
125.    
     
126.       
     
127.     /* 建立线程 */
     
128.     for (i = 0; i < sid_counter; i++) {
     
129.         info_p[i].file_start_num = 0;
     
130.         info_p[i].file_finish_num = 0;
     
131.         info_p[i].sid = i;
     
132.         pthread_create(&info_p[i].tid,
     
133.             NULL, receive_pthread, info_p + i);
     
134.     }
     
135. 
     
136.     for (i = 0; i < sid_counter; i++) {
     
137.             pthread_join(info_p[i].tid, &value_ptr);
     
138.     }
     
139.     pag_close();
     
140.     free(info_p);
     
141.     exit(1);
     
142. }
     

复制代码

snow888

1. 
   
2. /*
   
3. * write packet to the file
   
4. */
   
5. static void write_pkt(struct pktinfo *pktinfo_ptr)
   
6. {
   
7.     struct tcp_stream *tcp = pktinfo_ptr->pdata;
   
8.     FILE *fp = (FILE *)tcp->appinfo;
   
9.     struct half_stream *half_p = pag_gethalfstream(pktinfo_ptr);
   
10.     if (half_p == NULL) {
    
11.         return;
    
12.     }
    
13.    //写入负载
    
14.     if (half_p->count_new > 0) {
    
15.         fwrite(half_p->data, half_p->count_new, 1 , fp);
    
16.         fflush(fp);
    
17.     }
    
18.     fclose(fp);    // 关闭文件指针
    
19. }
    

复制代码

ddzhouqian

linux下一个进程可以打开的文件指针是有限制的

sparkzh

不用的文件要close啊

James_think

对一个文件处理完了，要记得关闭！

herocsz

贴了这么多源码，为什么不直接上伪代码呢？或者流程图?

lexken

本人是Linux C初学者，现在需要写一个实时读取加速网卡数据并将其负载写盘的程序。我现在写的程序实现4个线 ...
colrain_h 发表于 2011-11-03 21:24

    要有一個open之後,對應一個close.
如果是沒法知道是否open之後有關掉的.
好像可以用:
        fdn = getdtablesize();
        for(i=0; i<fdn; i++)
                close(i);