linux 下多线程文件传输程序，传输偏移量问题

//这是一个基于TCP协议的多线程文件传输程序（单线程已经实现），思想是：客户端先将要发送的文件分割成N个小文件
记录，然后启动N个线程发送，发送的数据信息格式是：偏移量+文件内容，服务器端将接收的内容解析，首先解析出偏移量
，转换成主机字节顺序后，根据此偏移量，把内容写入文件中，直到写入内容大小等于客户端先前发送文件的总大小（客户端
最开始把文件名和总文件大小发到服务器端）
现在的问题是：
1.这个偏移量发过去后，有的是对的，有的是错误的（表现为一个大的负数），我已经将字节顺序调整了的，
并且服务器端接收偏移量的类型off_t,在我的机子上是4个字节，足以容纳偏移量大小，请大家帮忙找出原因？
2.希望大家能够提出我程序中的毛病和很多不规范的地方，能使本程序更高效的运行

//filename:common.h
#ifndef _SERVER_H
#define _SERVER_H
#include <sys/socket.h>
#include <netinet/in.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <time.h>
#include <sys/time.h>
#include <errno.h>
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/select.h>
#include <time.h>
#include <pthread.h>
#include <signal.h>
#define SERVERPORT 8000
#define MAX_CONNECTION_NO 10
#define MAXBUFFER 1024*5
#define DEFAULTDIR "/mnt/hgfs/win"
#endif




//客户端代码

#include "common.h"
void doit(int fd,char* filename);
size_t writen(int fd,void *ptr,size_t n);
off_t get_file_size(char *name);
int connect_server(int port,char *s);//return connect socket
char* get_file_name(char *s);
int split_file(char *filename,long size);
//void thread_create(void);
void thread_wait(void);
int thread_num;
pthread_t thread[10];
static void * thread_run(void* arg);
#define SIZE 1024*1024//每个小文件大小

struct arg
{
   int fd;
   char filename[256];
};
int main(int argc,char **argv)
{
if(argc!=3)
{
printf("usage<./client ip filename>\n");
return;
}
int connfd;
    connfd=connect_server(SERVERPORT,argv[1]);
if(connfd<0) return 1;
doit(connfd,argv[2]);
close(connfd);
return 0;
}
int connect_server(int port,char *s)//return connect socket
{
int clifd;
clifd=socket(AF_INET,SOCK_STREAM,0);
if(clifd<0)
{
perror("socket() error:");
return -1;
}
struct sockaddr_in desadd;
desadd.sin_family=AF_INET;
desadd.sin_port=htons(SERVERPORT);
desadd.sin_addr.s_addr=inet_addr(s);
if(connect(clifd,(struct sockaddr*)&desadd,sizeof(struct sockaddr))<0)
{

perror("connect():");
return -1;
}
return clifd;
   
}
void doit(int clifd,char *filename)
{
char buffer[256];
int len;
int fd=open(filename,O_RDONLY);
if(fd<0)
{
perror("fopen() error:");
return;
}
int wlen;
memset(buffer,MAXBUFFER,0);
    int i;
    off_t filesize=get_file_size(filename);
    if((filename=get_file_name(filename))==NULL) return;
printf("文件大小为：%ld\n",filesize);
off_t fsize=htonl(filesize);
memcpy(buffer,&fsize,sizeof(fsize));
memcpy(buffer+sizeof(fsize),filename,strlen(filename));
time_t begin_time=time((time_t*)0);//开始时间
wlen=writen(clifd,buffer,sizeof(fsize)+strlen(filename));//发送文件大小
off_t position=0;
/*单线程传送文件485M文件花费131秒*/

char buf[256];//可以用动态内存的方式
thread_num=split_file(filename,SIZE);//分解成每份SIZE大小的小文件 返回文件数目
pthread_t tid;

memset(thread,0,sizeof(thread));
int tmp;
for(i=1;i<=thread_num;i++)//产生thread_num个线程
{
struct arg argment;
argment.fd=clifd;
sprintf(buf,"%d_%s",i,filename);
strcpy(argment.filename,buf);
tmp=pthread_create(&thread[i-1],NULL,&thread_run,(void*)&argment);
if(tmp!=0)
printf("线程%d创建失败\n",i);
sleep(1);
}
    thread_wait();
    time_t end_time=time((time_t*)0);
printf("take the time:%lf(s)\n",difftime(end_time,begin_time));
close(fd);
}
void thread_wait(void)
{
        /*等待线程结束*/
   int i;
   for(i=0;i<thread_num;i++)
        if(thread[i] !=0) {                   //comment4
                pthread_join(thread[i],NULL);
                printf("线程%d已经结束\n",thread[i]);
        }
       
}

/*线程运行函数*/
static void * thread_run(void* arg)
{
   char buffer[MAXBUFFER];
  // printf("thread %d coming\n",pthread_self());
   struct arg argment=(*(struct arg*)arg);
   int fd=open(argment.filename,O_RDONLY);
   if(fd<0)
   {
   printf("线程%d打开文件失败\n",pthread_self());
   return NULL;
   }
   off_t pos;
   int ret0;
   A:
    ret0=read(fd,buffer,sizeof(pos));//第一次读出偏移量
   if(ret0<0&&errno==EINTR)
   goto A;
   else if(ret0<0)
   {
    printf("线程%d读文件失败\n",pthread_self());
    return NULL;
   }
   off_t position=*((int*)buffer);
   printf("%s的偏移量为:%d\n",argment.filename,position);

   
   if(lseek(fd,0,SEEK_END)<0) 
   {
perror("lseek(fd,0,SEEK_END)");
return;
}
off_t end_position=lseek(fd,0,SEEK_CUR);//得到文件末尾的位置
off_t offset=0;
int ret;
int wlen;
   while(offset<end_position)
   {
 
      memset(buffer,0,MAXBUFFER);
  lseek(fd,offset,SEEK_SET);
  printf("该数据包的偏移量为:%d\n",position);
  position=htonl(position);
  
  memcpy(buffer,&position,sizeof(off_t));
  ret=read(fd,buffer+sizeof(off_t),MAXBUFFER);
  if(ret<0&&errno==EINTR) continue;
  else if(ret<0) 
  {
  perror("in thread read() error:");
  return NULL;
  }
  wlen=writen(argment.fd,buffer,ret+sizeof(offset));//发送过去
  printf("thread发送了wlen=%d\n",pthread_self(),wlen);
  offset+=ret;
  position+=ret;
   }
   return NULL;
}
int split_file(char *filename,long size)//size 表示每块的大小
{
char buffer[MAXBUFFER];
memset(buffer,0,MAXBUFFER);
    off_t filesize=get_file_size(filename);
//
int n=filesize/size;
int bulk_num=((n==0)?n:n+1);//分成这么多快 
int i=1;
char buf[256];
int readn,wlen;
int written=0;
off_t position=0;
int fd=open(filename,O_RDONLY);//打开总的文件
if(fd<0) perror("open error\n");
int ret;
if(lseek(fd,0,SEEK_END)<0) 
{
perror("lseek(fd,0,SEEK_END)");
return -1;
}
ret=lseek(fd,0,SEEK_CUR);
int first;
while(i<=bulk_num)
{
sprintf(buf,"%d_%s",i,filename);//行成分割后的文件名
first=1;
FILE* fp=fopen(buf,"a");
if(!fp)
{
perror("fopen() error:");
return -1;
}
while(written<size)
{
    if(position==ret) break;
if(lseek(fd,position,SEEK_SET)<0)
{
perror("lseek() error:\n");
return -1;
}
if(first==1)
{
fwrite(&position,sizeof(position),1,fp);//写入此文件的偏移量便于组装
printf("%s的文件偏移为：%d\n",buf,position);
}
first=0;
readn=read(fd,buffer,MAXBUFFER);
if(readn<0&&errno==EINTR) continue;
else if(readn<0)
{
perror("read() error:");
return -1;
}
    position+=(off_t)readn;
            wlen=fwrite(buffer,1,readn,fp);
fflush(fp);
memset(buffer,0,MAXBUFFER);
written+=wlen;
    

}
fclose(fp);//注意每次要关闭文件
i++;
        written=0;

}
close(fd);
return bulk_num;

}
off_t get_file_size(char *filename)
{
struct stat buf;
off_t filesize;
    if(stat(filename,&buf)<0)
{
perror("stat() error:");
filesize=0;
}
filesize=(long)buf.st_size;//字节单位
return filesize;
}
size_t readn(int fd,void *ptr,size_t n)
{
size_t left=n;
char *s=ptr;
size_t ret;
while(left>0)
{
if((ret=read(fd,s,left))<0)
{
if(errno==EINTR) continue;
else return -1;
}
else if(ret==0)
{
            break;/*EOF*/
}
left-=ret;
s+=ret;
}
return (n-left);
}
size_t writen(int fd,void *ptr,size_t n)
{
size_t left=n;
char *s=ptr;
size_t ret;
while(left>0)
{
if((ret=write(fd,s,left))<=0)
{
if(ret<0&&errno==EINTR)//被中断
ret=0;
else
return -1;
}
left-=ret;
s+=ret;
}
return n;
}

char *get_file_name(char *s)
{
  int len=strlen(s);
  len--;
  if(s[len]=='/')
  {
  printf("请选择一个文件!\n");
  return NULL;
  }
  len--;;
  while(len>0&&s[len]!='/')
     len--;
 if(len==0&&s[len]!='/') return s;
 else 
 return s+len+1;
}
///////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////

16 个解决方案

#1

//服务器端代码

#include "common.h"
void doit(int connfd);
void sig_child(int signo);
int main()
{
  int sockfd;
  sockfd=socket(AF_INET,SOCK_STREAM,0);
  
  if(sockfd<0)
  {
    perror("sockfd error:");
    return 1;
  }

  struct sockaddr_in server_addr;
  bzero(&server_addr,0);
  server_addr.sin_family=AF_INET;
  server_addr.sin_port=htons(SERVERPORT);
  server_addr.sin_addr.s_addr=htons(INADDR_ANY);


  if(bind(sockfd,(struct sockaddr*)&server_addr,sizeof(struct sockaddr))<0)
  {
    perror("bind error:");
    return 1;
  }

  if(listen(sockfd,MAX_CONNECTION_NO)<0)
  {
    perror("listen error");
    return 1;
  }
  int connfd;
  struct sockaddr_in cliaddr;
  int clilen=sizeof(struct sockaddr);
  fd_set fdset;
  FD_ZERO(&fdset);
  int max_fd=sockfd+1;
  struct timeval timeout;
 // timeout.tv_sec=2;
 // timeout.tv_usec=0;
  int retval;
  pid_t pid;
  if(signal(SIGCHLD,sig_child)<0)
  {
  perror("signal() error:");
  return 1;
  }
  while(1)
  {
     FD_SET(sockfd,&fdset);
     retval=select(max_fd,&fdset,NULL,NULL,NULL);
     if(retval<0)
     {
       if(errno=EINTR) continue;
       else
       {
          perror("select error:");
          return 1;
       }
     }
     else if(retval==0)
     {
        printf("timeout\n");
        continue;
     }
     if(FD_ISSET(sockfd,&fdset))
     {
        connfd=accept(sockfd,(struct sockaddr*)&cliaddr,&clilen);
if(connfd<0)
{
if(errno==EINTR) continue;
else 
{
 perror("accept():");
 return 1;
}
}
if((pid=fork())==0)//父进程应该回收子进程
{
close(sockfd);
doit(connfd);
close(connfd);
exit(0);
}
close(connfd);
      
     }
    
    
  }
}
void sig_child(int signo)
{
pid_t pid;
while((pid=waitpid(-1,NULL,WNOHANG))>0);
return;
}
void doit(int connfd)
{
  char buffer[MAXBUFFER];
  memset(buffer,MAXBUFFER,0);
 
  int wlen,ret;
  long filesize;
  long count=0;
  char filename[256];
  memset(filename,0,256);//这点很关键
  int i;
  X:
  if((ret=read(connfd,buffer,MAXBUFFER))>0)//第一次读 发送文件的长度和文件名
  {
   char ch[4];
   buffer[ret]=0;
   memcpy(ch,buffer,4);
   filesize=*((off_t*)ch);
       filesize=ntohl(filesize);//得到长度
        printf("接收到的文件长度为：%ld\n",filesize);

   // 分离出文件名
   memcpy(filename,buffer+4,ret-4);
       filename[strlen(filename)]=0;
   
   printf("接收到的文件名为：%s\n",filename);
  }
  if(ret<0&&errno==EINTR)
  goto X;
  char savedir[256];
  sprintf(savedir,"%s/%s",DEFAULTDIR,filename);
  FILE *fp=fopen(savedir,"w");//如果文件不存在则创建一个新文件
  if(!fp)
  {
  perror("fopen error:");
  return;
  }
  int receive_completed=0;
  off_t position;
  memset(buffer,0,sizeof(MAXBUFFER));
  Z:
  while((ret=read(connfd,buffer,MAXBUFFER))>0&&count<filesize/*!receive_completed*/)//用线程去实现 同时写一个文件 那么要用到文件锁？？（从不同的地方写要吗？？）
  {
      unsigned char ch[4];
  memset(ch,0,sizeof(ch));
  memcpy(ch,buffer,sizeof(off_t));
  position=*((off_t*)ch);
  position=ntohl(position);//得到偏移量
      printf("该数据包的偏移量为:%ld\n",position);

  /*如果偏移量不正确下面写文件肯定为错，故注释掉，否则会内存耗尽*/
    //fseek(fp,position,SEEK_SET);  
    //wlen=fwrite(buffer+sizeof(off_t),1,ret-sizeof(off_t),fp);//返回的是数据项的数目 注意啊
// fflush(fp);
    // count+=wlen;
  count+=(ret-sizeof(off_t));
      memset(buffer,MAXBUFFER,0);
  printf("已经接收 count=%d\n",count);
  }
  if(count>=filesize&&count!=0) printf("文件传送完毕\n");
  if(ret==0)
  {
    printf("client close\n");
  }
  else if(errno==EINTR) goto Z;
  else
  {
   perror("read eorr!");
   return;
  }
 
   fclose(fp);
}

#2

这个问题不是传送处理问题，是使用线程不当！

ret0=read(fd,buffer,sizeof(pos));//第一次读出偏移量 

例如当一个线程刚读入了偏移量头，（还没有读取文件数据），此时另一个线程便也试图读入偏移量头，此时它读入的就
不是下一帧数据的偏移量头，而是上一帧的文件数据．

下面是建议：
使用线程不能加快接收速度，ｓｏｃｋｅｔ还是串行传输的．
多线程主要用于完成并行的计算或处理，而不是Ｉ／Ｏ操作．

#3

to The final:
另一个线程始终是读的另一个文件的数据啊

#4

我没仔细看代码.
时使用一个socket连接吗?
如果是,就会是以上问题.

#5

是一个连接
但是启动线程时，他们读的不是一个文件，每个线程读不同的文件

#6

一个连接,他们的数据是串行传输的,不同的文件的不同部分都参杂在一起你如何区分?
把它当流来考虑你就明白了.

不仅读取/发送socket有问题, 就是写文件会有问题!
多线程考虑的必须很全面, 所有的操作是否是线程安全的?公共变量,资源是否是线程安全的?
多线程并行处理公共资源是否有逻辑问题等等.


 

#7

不同的文件通过偏移量来区分，写的时候根据偏移量来组合，这里我认为线程不需要同步，因为他们不是操作的同一资源，而是操作的不同文件

#8

一个连接,就是如下问题.

例如当一个线程刚读入了偏移量头，（还没有读取文件数据），此时另一个线程便也试图读入偏移量头，此时它读入的就 
不是下一帧数据的偏移量头，而是上一帧的文件数据

#9

我想明白了，那么用多线程应该怎么实现呢？

#10

。。你可以将读出来的东西先放在一个全局缓冲区内，然后开一个发送线程去发送缓冲区的东西。。。注意发送
线程和读取线程的同步就可以了

#11

引用 9 楼 shunzi__1984 的回复:

我想明白了，那么用多线程应该怎么实现呢？

采用多线程要针对它能解决什么问题, 一两句说不清,建议看看多线程开发书.
不要被多线程下载程序如flashget等表面现象蒙蔽了, 表面看到的是多连接socket.

#12

我想到一种解决方案:
  每个线程发送的时候，先发送要发送内容的长度，再发送该内容在原文件中的偏移量+文件内容（这两个必须是原子操作）
   pthread_mutex_lock(&mut);
  writen(argment.fd,&packet_len,sizeof(int));//先写偏移量
  wlen=writen(argment.fd,buffer,ret+sizeof(offset));//发送过去
  pthread_mutex_unlock(&mut);
  用以上几条代码，应该能保证吧

接收端然后根据接收到的长度，接收数据包
这样应该没有问题吧？

但是我调试时，比以前的问题更糟？不仅接收端读偏移量有大负数，发送段线程读偏移量时也出现了负数

为什么呢？

#13

高手不少啊！！

#14

hao 

#15

非常好

#16

哎，怎么最后也没有解决啊