/*
* lftp - file transfer program
*
* Copyright (c) 1996-2017 by Alexander V. Lukyanov (lav@yars.free.net)
*
* This program is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation; either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
/* FileCopyPeer behaviour:
1) when suspended, does nothing
2) tries to read some data at seek_pos, sets pos to position of Get (get).
2.5) tries to position to seek_pos and gets ready to write (put).
3) if it cannot seek to seek_pos, changes pos to what it can seek.
4) if it knows that it cannot seek to pos>0, CanSeek()==false
5) if it knows that it cannot seek to pos==0, CanSeek0()==false
6) it tries to get date/size if told to. (get)
7) it sets date on the file if eof is reached and date is known (put).
8) if put needs size/date before it writes data, NeedSizeDateBeforehand()==true.
*/
#include <config.h>
#include <assert.h>
#include <unistd.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <math.h>
#include <stddef.h>
#include "FileCopy.h"
#include "url.h"
#include "log.h"
#include "misc.h"
#include "LsCache.h"
#include "plural.h"
#include "ArgV.h"
#define skip_threshold 0x1000
ResDecl rate_period ("xfer:rate-period","15", ResMgr::UNumberValidate,ResMgr::NoClosure);
ResDecl eta_period ("xfer:eta-period", "120",ResMgr::UNumberValidate,ResMgr::NoClosure);
ResDecl max_redir ("xfer:max-redirections", "5",ResMgr::UNumberValidate,ResMgr::NoClosure);
ResDecl buffer_size ("xfer:buffer-size","0x10000",ResMgr::UNumberValidate,ResMgr::NoClosure);
// It's bad when lftp receives data in small chunks, try to accumulate
// data in a kernel buffer using a delay and slurp it at once:
enum {
// Delays in microseconds
MAX_DELAY=30000,
DELAY_STEP=30,
// This size is related to socket buffer size.
// When it is too large, tcp slowdown happens.
// SSL has packet size 0x4000, so we have to use a lower threshold.
MAX_READ_TO_DELAY=0x3F00,
};
// FileCopy
#define super SMTask
#define set_state(s) do { state=(s); \
Log::global->Format(11,"FileCopy(%p) enters state %s\n", this, #s); } while(0)
int FileCopy::Do()
{
int m=STALL;
const char *b;
int s;
int rate_add;
if(Error() || Done())
return m;
switch(state)
{
pre_INITIAL:
set_state(INITIAL);
m=MOVED;
case(INITIAL):
if(remove_target_first && !put->FileRemoved())
return m;
remove_target_first=false;
if(cont && put->CanSeek())
put->WantSize();
if(put->NeedSizeDateBeforehand() || (cont && put->CanSeek() && put->GetSize()==NO_SIZE_YET))
{
if(get->GetSize()==NO_SIZE_YET || get->GetDate()==NO_DATE_YET)
{
put->Suspend();
get->DontStartTransferYet();
get->Resume();
get->WantSize();
if(put->NeedDate())
get->WantDate();
goto pre_GET_INFO_WAIT;
}
}
if(get->GetSize()==NO_SIZE_YET)
get->WantSize();
if(get->GetSize()!=NO_SIZE && get->GetSize()!=NO_SIZE_YET)
put->SetEntitySize(get->GetSize());
if(get->GetDate()!=NO_DATE && get->GetDate()!=NO_DATE_YET)
put->SetDate(get->GetDate());
else if(get->GetDate()==NO_DATE_YET)
{
if(put->NeedDate())
get->WantDate();
}
if(cont && put->CanSeek())
put->Seek(FILE_END);
else
{
if(put->range_start>0 && put->CanSeek())
put->Seek(put->range_start);
if(get->range_start>0 && get->CanSeek())
get->Seek(get->range_start);
goto pre_DO_COPY;
}
get->Suspend();
put->Resume();
set_state(PUT_WAIT);
m=MOVED;
/* fallthrough */
case(PUT_WAIT):
if(put->Error())
goto put_error;
if(put->GetSeekPos()!=FILE_END && get->GetSize()>=0
&& put->GetSeekPos()>=get->GetSize())
{
debug((9,_("copy: destination file is already complete\n")));
if(get->GetDate()!=NO_DATE)
goto pre_CONFIRM_WAIT; // have to set the date.
goto pre_GET_DONE_WAIT;
}
if(!put->IOReady())
return m;
/* now we know if put's seek failed. Seek get accordingly. */
if(get->CanSeek())
get->Seek(put->GetRealPos());
pre_DO_COPY:
get->Resume();
get->StartTransfer();
RateReset();
set_state(DO_COPY);
m=MOVED;
/* fallthrough */
case(DO_COPY): {
if(put->Error())
{
put_error:
SetError(put->ErrorText());
return MOVED;
}
if(get->Error() && get->Size()==0)
{
put->DontVerify();
if(put->GetPos()>0)
{
put->PutEOF();
put->Roll();
}
get_error:
SetError(get->ErrorText());
return MOVED;
}
if(put->Broken())
{
get->Suspend();
if(!put->Done())
return m;
debug((9,_("copy: put is broken\n")));
if(fail_if_broken)
{
SetError(strerror(EPIPE));
return MOVED;
}
goto pre_GET_DONE_WAIT;
}
put->Resume();
if(put->GetSeekPos()==FILE_END) // put position is not known yet.
{
get->Suspend();
return m;
}
get->Resume();
if(fail_if_cannot_seek && (get->GetRealPos()<get->range_start
|| put->GetRealPos()<put->range_start
|| get->GetRealPos()!=put->GetRealPos()))
{
SetError(_("seek failed"));
return MOVED;
}
if(high_watermark_timeout.Stopped())
{
SetError(_("no progress timeout"));
return MOVED;
}
if(get->GetSize()>0 && get->GetRealPos()>get->GetSize())
{
get->SetSize(NO_SIZE_YET);
get->SetDate(NO_DATE_YET);
}
long lbsize=0;
if(line_buffer)
lbsize=line_buffer->Size();
/* check if positions are correct */
off_t get_pos=get->GetRealPos()-get->range_start;
off_t put_pos=put->GetRealPos()-put->range_start;
if(get_pos-lbsize!=put_pos)
{
if(line_buffer)
line_buffer->Empty();
if(get_pos==put_pos)
{ // rare case.
return MOVED;
}
if(put_pos<get_pos)
{
if(!get->CanSeek(put->GetRealPos()))
{
// we lose... How about a large buffer?
SetError(_("cannot seek on data source"));
return MOVED;
}
debug((9,_("copy: put rolled back to %lld, seeking get accordingly\n"),
(long long)put->GetRealPos()));
debug((10,"copy: get position was %lld\n",
(long long)get->GetRealPos()));
get->Seek(put->GetRealPos());
return MOVED;
}
else // put_pos > get_pos
{
off_t size=get->GetSize();
if(size>=0 && put->GetRealPos()>=size)
{
// simulate eof, as we have already have the whole file.
debug((9,_("copy: all data received, but get rolled back\n")));
goto eof;
}
off_t skip=put->GetRealPos()-get->GetRealPos();
if(!put->CanSeek(get->GetRealPos()) || skip<skip_threshold)
{
// we have to skip some data
get->Get(&b,&s);
if(skip>s)
skip=s;
if(skip==0)
return m;
get->Skip(skip);
bytes_count+=skip;
return MOVED;
}
debug((9,_("copy: get rolled back to %lld, seeking put accordingly\n"),
(long long)get->GetRealPos()));
put->Seek(get->GetRealPos());
return MOVED;
}
}
if(put->IsFull())
get->Suspend(); // stall the get.
get->Get(&b,&s);
if(b==0) // eof
{
debug((10,"copy: get hit eof\n"));
goto eof;
}
rate_add=put_buf;
if(s==0)
{
put_buf=put->Buffered();
rate_add-=put_buf;
RateAdd(rate_add);
if(put->Size()==0)
put->Suspend();
return m;
}
m=MOVED;
if(get->range_limit!=FILE_END && get->range_limit<get->GetRealPos()+s)
{
s=get->range_limit-get->GetRealPos();
if(s<0)
s=0;
}
if(line_buffer)
{
const char *lb;
int ls;
if(line_buffer->Size()>line_buffer_max)
{
line_buffer->Get(&lb,&ls);
put->Put(lb,ls);
line_buffer->Skip(ls);
}
line_buffer->Put(b,s);
get->Skip(s);
bytes_count+=s;
// now find eol in line_buffer.
line_buffer->Get(&lb,&ls);
const char *eol=0;
if(get->Eof() || get->Error())
eol=lb+ls-1;
else
eol=memrchr(lb,'\n',ls);
if(eol)
{
put->Put(lb,eol-lb+1);
line_buffer->Skip(eol-lb+1);
}
}
else
{
put->Put(b,s);
get->Skip(s);
bytes_count+=s;
}
put_buf=put->Buffered();
rate_add-=put_buf-s;
RateAdd(rate_add);
if(high_watermark<put_pos+s)
{
high_watermark=put_pos+s;
high_watermark_timeout.Reset();
}
if(get->range_limit!=FILE_END && get->range_limit<=get->GetRealPos())
{
debug((10,"copy: get reached range limit\n"));
goto eof;
}
return m;
}
eof:
if(line_buffer)
{
line_buffer->Get(&b,&s);
put->Put(b,s);
line_buffer->Skip(s);
}
if(!CheckFileSizeAtEOF())
{
SetError(_("file size decreased during transfer"));
return MOVED;
}
pre_CONFIRM_WAIT:
put->SetSuggestedFileName(get->GetSuggestedFileName());
put->SetDate(get->GetDate());
if(get->GetSize()!=NO_SIZE && get->GetSize()!=NO_SIZE_YET)
put->SetEntitySize(get->GetSize());
put->PutEOF();
get->Suspend();
put->Resume();
put_eof_pos=put->GetRealPos();
debug((10,"copy: waiting for put confirmation\n"));
set_state(CONFIRM_WAIT);
m=MOVED;
case(CONFIRM_WAIT):
if(put->Error())
goto put_error;
/* check if put position is correct */
if(put_eof_pos!=put->GetRealPos() || put->GetSeekPos()==FILE_END)
{
set_state(DO_COPY);
return MOVED;
}
rate_add=put_buf;
put_buf=put->Buffered();
rate_add-=put_buf;
RateAdd(rate_add);
if(!put->Done())
return m;
debug((10,"copy: put confirmed store\n"));
pre_GET_DONE_WAIT:
get->Empty();
get->PutEOF();
get->Resume();
set_state(GET_DONE_WAIT);
m=MOVED;
end_time=now;
put->Suspend();
/* fallthrough */
case(GET_DONE_WAIT):
if(get->Error())
goto get_error;
if(remove_source_later)
{
get->RemoveFile();
remove_source_later=false;
}
if(!get->Done())
return m;
debug((10,"copy: get is finished - all done\n"));
set_state(ALL_DONE);
get->Suspend();
LogTransfer();
return MOVED;
pre_GET_INFO_WAIT:
set_state(GET_INFO_WAIT);
m=MOVED;
case(GET_INFO_WAIT):
if(get->Error())
goto get_error;
if(get->GetSize()==NO_SIZE_YET || get->GetDate()==NO_DATE_YET)
return m;
goto pre_INITIAL;
case(ALL_DONE):
return m;
}
return m;
}
FileCopy::FileCopy(FileCopyPeer *s,FileCopyPeer *d,bool c)
: get(s), put(d), cont(c),
rate("xfer:rate-period"),
rate_for_eta("xfer:eta-period"),
high_watermark_timeout("xfer:timeout",0)
{
set_state(INITIAL);
int max_buf=buffer_size.Query(0);
if(max_buf<1)
max_buf=1;
s->SetMaxBuffered(max_buf);
d->SetMaxBuffered(max_buf);
put_buf=0;
put_eof_pos=0;
high_watermark=0;
bytes_count=0;
fail_if_cannot_seek=false;
fail_if_broken=true;
remove_source_later=false;
remove_target_first=false;
line_buffer_max=0;
}
FileCopy::~FileCopy()
{
}
FileCopy *FileCopy::New(FileCopyPeer *s,FileCopyPeer *d,bool c)
{
FileCopy *res=0;
if(fxp_create)
res=fxp_create(s,d,c);
if(res)
return res;
return new FileCopy(s,d,c);
}
void FileCopy::SuspendInternal()
{
super::SuspendInternal();
if(get) get->SuspendSlave();
if(put) put->SuspendSlave();
}
void FileCopy::ResumeInternal()
{
if(get) get->ResumeSlave();
if(put) put->ResumeSlave();
super::ResumeInternal();
}
void FileCopy::Fg()
{
if(get) get->Fg();
if(put) put->Fg();
}
void FileCopy::Bg()
{
if(get) get->Bg();
if(put) put->Bg();
}
void FileCopy::SetError(const char *str)
{
error_text.set(str);
get=0;
}
void FileCopy::LineBuffered(int s)
{
if(!line_buffer)
line_buffer=new Buffer();
line_buffer_max=s;
}
off_t FileCopy::GetPos() const
{
if(put) {
off_t pos = put->GetRealPos() - put->Buffered();
// sometimes Buffered overestimates the amount of buffered data
if(pos<0)
pos=0;
return pos;
}
if(get)
return get->GetRealPos();
return 0;
}
off_t FileCopy::GetSize() const
{
if(get)
return get->GetSize();
return NO_SIZE;
}
int FileCopy::GetPercentDone() const
{
if(!get || !put)
return 100;
off_t size=get->GetSize();
if(size==NO_SIZE || size==NO_SIZE_YET)
return -1;
if(size==0)
return 0;
off_t ppos=put->GetRealPos() - put->Buffered() - put->range_start;
if(ppos<0)
return 0;
off_t psize=size-put->range_start;
if(put->range_limit!=FILE_END)
psize=put->range_limit-put->range_start;
if(psize<0)
return 100;
if(ppos>psize)
return -1;
return percent(ppos,psize);
}
const char *FileCopy::GetPercentDoneStr() const
{
int pct=GetPercentDone();
if(pct==-1)
return "";
static char buf[8];
snprintf(buf,8,"(%d%%) ",pct);
return buf;
}
void FileCopy::RateAdd(int a)
{
rate.Add(a);
rate_for_eta.Add(a);
}
void FileCopy::RateReset()
{
start_time=now;
rate.Reset();
rate_for_eta.Reset();
}
float FileCopy::GetRate()
{
if(!rate.Valid() || !put)
return 0;
return rate.Get();
}
const char *FileCopy::GetRateStr()
{
if(!rate.Valid() || !put)
return "";
return rate.GetStrS();
}
off_t FileCopy::GetBytesRemaining()
{
if(!get)
return 0;
if(get->range_limit==FILE_END)
{
off_t size=get->GetSize();
if(size<=0 || size<get->GetRealPos() || !rate_for_eta.Valid())
return -1;
return(size-GetPos());
}
return get->range_limit-GetPos();
}
const char *FileCopy::GetETAStr()
{
off_t b=GetBytesRemaining();
if(b<0 || !put)
return "";
return rate_for_eta.GetETAStrSFromSize(b);
}
long FileCopy::GetETA(off_t b)
{
if(b<0 || !rate_for_eta.Valid())
return -1;
return (long)(double(b) / rate_for_eta.Get() + 0.5);
}
const char *FileCopy::GetStatus()
{
static xstring buf;
const char *get_st=get?get->GetStatus():0;
const char *put_st=put?put->GetStatus():0;
if(get_st && get_st[0] && put_st && put_st[0])
buf.vset("[",get_st,"->",put_st,"]",NULL);
else if(get_st && get_st[0])
buf.vset("[",get_st,"]",NULL);
else if(put_st && put_st[0])
buf.vset("[",put_st,"]",NULL);
else
return "";
return buf;
}
double FileCopy::GetTimeSpent()
{
if(end_time<start_time)
return 0;
return TimeDiff(end_time,start_time);
}
FgData *FileCopy::GetFgData(bool fg)
{
// NOTE: only one of get/put can have FgData in this implementation.
FgData *f=0;
if(get) f=get->GetFgData(fg);
if(f) return f;
if(put) f=put->GetFgData(fg);
return f;
}
pid_t FileCopy::GetProcGroup()
{
pid_t p=0;
if(get) p=get->GetProcGroup();
if(p) return p;
if(put) p=put->GetProcGroup();
return p;
}
void FileCopy::Kill(int sig)
{
if(get) get->Kill(sig);
if(put) put->Kill(sig);
}
Ref<Log> FileCopy::transfer_log;
void FileCopy::LogTransfer()
{
const char *log_ctx="xfer";
if(!ResMgr::QueryBool("log:enabled",log_ctx))
return;
const char *src=get->GetURL();
if(!src)
return;
src=alloca_strdup(src);
const char *dst=put->GetURL();
if(!dst)
return;
dst=alloca_strdup(dst);
if(!transfer_log)
transfer_log=new Log(log_ctx);
long long range_limit=GetRangeLimit();
if(range_limit==FILE_END)
range_limit=get->GetPos();
transfer_log->Format(0,"%s -> %s %lld-%lld %s\n",
url::remove_password(src),url::remove_password(dst),
(long long)GetRangeStart(),range_limit,
Speedometer::GetStrProper(GetBytesCount()/GetTimeSpent()).get());
}
void FileCopy::SetRange(off_t s,off_t lim)
{
get->SetRange(s,lim);
put->SetRange(s,lim);
}
bool FileCopy::CheckFileSizeAtEOF() const
{
long long range_limit=GetRangeLimit();
if(range_limit==FILE_END)
{
const long long size=GetSize();
if(size==NO_SIZE || size==NO_SIZE_YET)
return true; // nothing to compare with.
range_limit=size;
}
const long long get_pos=get->GetRealPos();
const long long put_pos=put->GetRealPos();
const long long pos=(get_pos>put_pos ? get_pos : put_pos);
if(pos<=0 || pos>=range_limit)
return true;
debug((0,"expected pos=%lld, actual pos=%lld\n",range_limit,pos));
return false;
}
// FileCopyPeer implementation
#undef super
#define super Buffer
off_t FileCopyPeer::GetSize()
{
if(size>=0 && pos>size)
WantSize();
return size;
}
void FileCopyPeer::SetSize(off_t s)
{
size=s;
if(seek_pos==FILE_END)
{
if(size!=NO_SIZE && size!=NO_SIZE_YET)
seek_pos=size;
else
seek_pos=0;
}
}
void FileCopyPeer::SetDate(time_t d,int p)
{
date.set(d,p);
if(d==NO_DATE || d==NO_DATE_YET)
date_set=true;
else
date_set=false;
}
void FileCopyPeer::SetRange(const off_t s,const off_t lim)
{
range_start=s;
range_limit=lim;
if(mode==PUT || range_start>GetPos()+0x4000)
Seek(range_start);
}
bool FileCopyPeer::Done()
{
if(Error())
return true;
if(eof && Size()==0)
{
if(removing)
return false;
if(mode==PUT)
return done;
return true;
}
if(broken)
return true;
return false;
}
void FileCopyPeer::Seek(off_t offs)
{
seek_pos=offs;
if(mode==PUT)
pos-=Size();
Empty();
eof=false;
broken=false;
}
const char *FileCopy::TempFileName(const char *file)
{
if(!ResMgr::QueryBool("xfer:use-temp-file",0))
return file;
xstring &temp=xstring::get_tmp(ResMgr::Query("xfer:temp-file-name",0));
if(temp.length()==0 || temp.eq("*"))
return file;
const char *name=basename_ptr(file);
int subst_pos=temp.instr('*');
if(subst_pos>=0)
temp.set_substr(subst_pos,1,name);
else {
if(temp.last_char()=='.')
temp.append(name);
else if(temp[0]=='.')
temp.set_substr(0,0,name);
else
temp.append('.').append(name);
}
return dir_file(dirname(file),temp);
}
const char *FileCopyPeer::UseTempFile(const char *file)
{
const char *temp=FileCopy::TempFileName(file);
if(temp==file)
return file;
auto_rename=true;
temp_file=true;
SetSuggestedFileName(basename_ptr(file));
return temp;
}
FileCopyPeer::FileCopyPeer(dir_t m) : IOBuffer(m)
{
want_size=false;
want_date=false;
start_transfer=true;
size=NO_SIZE_YET;
date=NO_DATE_YET;
e_size=NO_SIZE;
seek_pos=0;
can_seek=false;
can_seek0=false;
date_set=false;
do_set_date=true;
do_verify=true;
ascii=false;
range_start=0;
range_limit=FILE_END;
removing=false;
file_removed=false;
temp_file=false;
do_mkdir=false;
use_cache=true;
write_allowed=true;
done=false;
auto_rename=false;
Suspend(); // don't do anything too early
}
// FileCopyPeerFA implementation
#undef super
#define super FileCopyPeer
int FileCopyPeerFA::Do()
{
int m=STALL;
int res;
if(session->OpenMode()==FA::MAKE_DIR)
{
// doing mkdir
int res=session->Done();
if(res==FA::IN_PROGRESS)
return m;
if(res<0)
debug((3,"mkdir failed: %s\n",session->StrError(res)));
session->Close();
m=MOVED;
}
else if(session->OpenMode()==FA::RENAME)
{
int res=session->Done();
if(res==FA::IN_PROGRESS)
return m;
if(res<0) {
if(temp_file)
SetError(session->StrError(res));
else
debug((3,"rename failed: %s\n",session->StrError(res)));
}
session->Close();
done=true;
return MOVED;
}
if(do_mkdir) {
// do mkdir just once
do_mkdir=false;
assert(!session->IsOpen());
const xstring& dir=dirname(file);
if(dir.length()>0 && dir.ne("/") && dir.ne(".") && dir.ne("..")) {
// FIXME: .././.. should be also excluded
session->Mkdir(dirname(file),true);
return MOVED;
}
}
if(removing)
{
res=session->Done();
if(res<=0)
{
removing=false;
file_removed=true;
session->Close();
Suspend();
m=MOVED;
}
return m;
}
if(Done() || Error())
return m;
if(verify)
{
if(verify->Error())
{
SetError(verify->ErrorText());
m=MOVED;
}
else if(verify->Done())
{
if(suggested_filename && auto_rename)
{
const char *new_name=dir_file(dirname(file),suggested_filename);
bool clobber=temp_file;
session->Rename(file,new_name,clobber);
return MOVED;
}
done=true;
m=MOVED;
}
return m;
}
// if we need some info and cannot start the transfer (yet),
// then use ARRAY_INFO to fetch the file information.
if(((want_size && size==NO_SIZE_YET) || (want_date && date==NO_DATE_YET))
&& (mode==PUT || !start_transfer) && session->IsClosed())
{
FileInfo *fi=new FileInfo(file);
if(want_size)
fi->Need(fi->SIZE);
if(want_date)
fi->Need(fi->DATE);
info.Empty();
info.Add(fi);
session->GetInfoArray(&info);
m=MOVED;
}
if(session->OpenMode()==FA::ARRAY_INFO)
{
res=session->Done();
if(res==FA::IN_PROGRESS)
return m;
if(res<0)
{
session->Close();
SetSize(NO_SIZE);
SetDate(NO_DATE);
return MOVED;
}
FileInfo *fi=info[0];
if(want_size)
SetSize(fi->size);
if(want_date)
SetDate(fi->date);
session->Close();
return MOVED;
}
switch(mode)
{
case PUT:
if(fxp)
{
if(eof)
goto fxp_eof;
return m;
}
res=Put_LL(buffer+buffer_ptr,Size());
if(res>0)
{
buffer_ptr+=res;
m=MOVED;
}
else if(res<0)
return MOVED;
if(Size()==0)
{
if(eof)
{
if(date!=NO_DATE && date!=NO_DATE_YET)
session->SetDate(date);
if(e_size!=NO_SIZE && e_size!=NO_SIZE_YET)
session->SetSize(e_size);
res=session->StoreStatus();
if(res==FA::OK)
{
session->Close();
fxp_eof:
// FIXME: set date for real.
date_set=true;
if(!verify && do_verify)
verify=new FileVerificator(session,file);
else
done=true;
return MOVED;
}
else if(res==FA::IN_PROGRESS)
return m;
else
{
if(res==FA::DO_AGAIN)
return m;
if(res==FA::STORE_FAILED)
{
upload_state.Save(session);
session->Close();
if(can_seek && seek_pos>0)
Seek(FILE_END);
else
Seek(0);
return MOVED;
}
SetError(session->StrError(res));
return MOVED;
}
return m;
}
}
break;
case GET:
if(eof)
return m;
if(fxp)
return m;
res=TuneGetSize(Get_LL(get_size));
if(res>0)
{
EmbraceNewData(res);
SaveMaxCheck(0);
return MOVED;
}
if(res<0)
return MOVED;
if(eof)
{
session->Close();
return MOVED;
}
break;
}
return m;
}
bool FileCopyPeerFA::IOReady()
{
if(seek_pos==0)
return true;
if(seek_pos==FILE_END && size==NO_SIZE_YET)
return false;
return session->IOReady();
}
void FileCopyPeerFA::SuspendInternal()
{
if(fxp && mode==PUT)
return;
if(session->IsOpen())
session->SuspendSlave();
super::SuspendInternal();
}
void FileCopyPeerFA::ResumeInternal()
{
super::ResumeInternal();
session->ResumeSlave();
}
const char *FileCopyPeerFA::GetStatus()
{
if(verify)
return verify->Status();
if(!session->IsOpen())
return 0;
return session->CurrentStatus();
}
void FileCopyPeerFA::Seek(off_t new_pos)
{
if(pos==new_pos)
return;
super::Seek(new_pos);
session->Close();
if(seek_pos==FILE_END)
WantSize();
else
pos=new_pos;
}
void FileCopyPeerFA::OpenSession()
{
current->Timeout(0); // mark it MOVED.
if(mode==GET)
{
if(size!=NO_SIZE && size!=NO_SIZE_YET && !ascii
&& (seek_pos>size || (seek_pos==size && size>0)))
{
past_eof:
debug((10,"copy src: seek past eof (seek_pos=%lld, size=%lld)\n",
(long long)seek_pos,(long long)size));
pos=seek_pos;
eof=true;
return;
}
const char *b;
int s;
int err;
if(use_cache && FileAccess::cache->Find(session,file,FAmode,&err,&b,&s))
{
if(err)
{
SetError(b);
return;
}
size=s;
if(seek_pos>=s)
goto past_eof;
b+=seek_pos;
s-=seek_pos;
Save(0);
Put(b,s);
pos=seek_pos;
eof=true;
return;
}
}
else // mode==PUT
{
if(e_size>=0 && size>=0 && seek_pos>=e_size)
{
debug((10,"copy dst: seek past eof (seek_pos=%lld, size=%lld)\n",
(long long)seek_pos,(long long)e_size));
eof=true;
if(date==NO_DATE || date==NO_DATE_YET)
return;
}
}
session->Open(file,FAmode,seek_pos);
session->SetFileURL(orig_url);
session->SetLimit(range_limit);
if(mode==PUT) {
upload_state.Restore(session);
if(e_size!=NO_SIZE && e_size!=NO_SIZE_YET)
session->SetSize(e_size);
if(date!=NO_DATE && date!=NO_DATE_YET)
session->SetDate(date);
} else {
if(size!=NO_SIZE && size!=NO_SIZE_YET)
session->SetSize(size);
}
session->RereadManual();
if(ascii)
session->AsciiTransfer();
if(want_size && size==NO_SIZE_YET)
session->WantSize(&size);
if(want_date && (date==NO_DATE_YET || date.ts_prec>0))
session->WantDate(&date);
if(mode==GET)
SaveRollback(seek_pos);
else
pos=seek_pos+Size();
}
void FileCopyPeerFA::WantSize()
{
struct stat st;
if(!strcmp(session->GetProto(),"file")
&& stat(dir_file(session->GetCwd(),file),&st)!=-1)
SetSize(S_ISREG(st.st_mode)?st.st_size:NO_SIZE);
else
super::WantSize();
}
void FileCopyPeerFA::RemoveFile()
{
session->Open(file,FA::REMOVE);
removing=true;
}
int FileCopyPeerFA::Get_LL(int len)
{
if(get_delay>0)
{
if(!get_ll_timer.Stopped())
return 0;
session->ResumeSlave();
}
int res=0;
if(session->IsClosed())
OpenSession();
if(eof) // OpenSession can set eof=true.
return 0;
off_t io_at=pos;
if(GetRealPos()!=io_at) // GetRealPos can alter pos.
return 0;
res=session->Read(this,len);
if(res<0)
{
if(res==FA::DO_AGAIN)
return 0;
if(res==FA::FILE_MOVED)
{
// handle redirection.
assert(!fxp);
const char *loc_c=session->GetNewLocation();
int max_redirections=max_redir.Query(0);
if(loc_c && loc_c[0] && max_redirections>0)
{
Log::global->Format(3,_("copy: received redirection to `%s'\n"),loc_c);
if(++redirections>max_redirections)
{
SetError(_("Too many redirections"));
return -1;
}
if(FAmode==FA::QUOTE_CMD)
FAmode=FA::RETRIEVE;
xstring loc(loc_c);
session->Close(); // loc_c is no longer valid.
loc_c=0;
ParsedURL u(loc,true);
if(u.proto)
{
my_session=FileAccess::New(&u);
session=my_session;
file.set(u.path?u.path.get():"");
orig_url.set(loc);
}
else // !proto
{
if(orig_url)
{
int p_ind=url::path_index(orig_url);
const char *s=strrchr(orig_url,'/');
int s_ind=s?s-orig_url:-1;
if(p_ind==-1 || s_ind==-1 || s_ind<p_ind)
s_ind=p_ind=strlen(orig_url);
if(loc[0]=='/')
{
orig_url.truncate(p_ind);
orig_url.append(loc);
}
else
{
orig_url.truncate(s_ind);
orig_url.append('/');
orig_url.append(loc);
}
}
loc.url_decode();
const char *slash=strrchr(file,'/');
if(loc[0]!='/' && slash)
{
file.truncate(slash-file);
file.set(dir_file(file,loc));
}
else
file.set(loc);
}
if(want_size || size!=NO_SIZE)
WantSize();
if(want_date || date!=NO_DATE)
WantDate();
upload_state.Clear();
current->Timeout(0); // retry with new location.
return 0;
}
}
SetError(session->StrError(res));
return -1;
}
else if(res==0)
{
debug((10,"copy-peer: EOF on %s\n",session->GetFileURL(session->GetFile()).get()));
eof=true;
FileAccess::cache->Add(session,file,FAmode,FA::OK,this);
SetSuggestedFileName(session->GetSuggestedFileName());
session->Close();
}
else if(res<=MAX_READ_TO_DELAY)
{
if(get_delay<=MAX_DELAY-DELAY_STEP)
get_delay+=DELAY_STEP;
get_ll_timer.SetMicroSeconds(get_delay);
session->SuspendSlave();
}
else if(res>MAX_READ_TO_DELAY && get_delay>=DELAY_STEP)
get_delay-=DELAY_STEP;
return res;
}
int FileCopyPeerFA::Put_LL(const char *buf,int len)
{
if(do_mkdir)
return 0; // can't write yet
if(session->IsClosed())
OpenSession();
off_t io_at=pos; // GetRealPos can alter pos, save it.
if(GetRealPos()!=io_at)
return 0;
if(len==0 && eof)
return 0;
int res=session->Write(buf,len);
if(res<0)
{
if(res==FA::DO_AGAIN)
return 0;
if(res==FA::STORE_FAILED)
{
upload_state.Save(session);
session->Close();
if(can_seek && seek_pos>0)
Seek(FILE_END);
else
Seek(0);
return 0;
}
SetError(session->StrError(res));
return -1;
}
seek_pos+=res; // mainly to indicate that there was some output.
return res;
}
int FileCopyPeerFA::PutEOF_LL()
{
if(mode==GET && session)
session->Close();
return 0;
}
off_t FileCopyPeerFA::GetRealPos()
{
if(session->OpenMode()!=FAmode || fxp)
return pos;
if(mode==PUT)
{
if(pos-Size()!=session->GetPos())
{
Empty();
can_seek=false;
pos=session->GetPos();
}
}
else
{
if(eof)
return pos;
if(session->GetRealPos()==0 && session->GetPos()>0)
{
can_seek=false;
session->SeekReal();
}
if(pos+Size()!=session->GetPos())
SaveRollback(session->GetPos());
}
return pos;
}
void FileCopyPeerFA::Init()
{
get_delay=0;
fxp=false;
redirections=0;
can_seek=true;
can_seek0=true;
if(FAmode==FA::LIST || FAmode==FA::LONG_LIST)
Save(FileAccess::cache->SizeLimit());
if(mode==PUT)
file.set(UseTempFile(file));
}
FileCopyPeerFA::FileCopyPeerFA(FileAccess *s,const char *f,int m)
: FileCopyPeer(m==FA::STORE ? PUT : GET), file(f),
my_session(s), session(my_session), FAmode(m)
{
Init();
}
FileCopyPeerFA::FileCopyPeerFA(const FileAccessRef& s,const char *f,int m)
: FileCopyPeer(m==FA::STORE ? PUT : GET), file(f),
session(s), FAmode(m)
{
Init();
}
FileCopyPeerFA::FileCopyPeerFA(const ParsedURL *u,int m)
: FileCopyPeer(m==FA::STORE ? PUT : GET), file(u->path), orig_url(u->orig_url),
my_session(FileAccess::New(u)), session(my_session), FAmode(m)
{
Init();
if(!file)
SetError(_("file name missed in URL"));
}
void FileCopyPeerFA::PrepareToDie()
{
if(session)
session->Close();
}
FileCopyPeerFA::~FileCopyPeerFA() {}
FileCopyPeerFA *FileCopyPeerFA::New(FileAccess *s,const char *url,int m)
{
ParsedURL u(url,true);
if(u.proto)
{
SessionPool::Reuse(s);
return new FileCopyPeerFA(&u,m);
}
return new FileCopyPeerFA(s,url,m);
}
FileCopyPeerFA *FileCopyPeerFA::New(const FileAccessRef& s,const char *url,int m)
{
ParsedURL u(url,true);
if(u.proto)
return new FileCopyPeerFA(&u,m);
return new FileCopyPeerFA(s,url,m);
}
FileCopyPeer *FileCopyPeerFA::Clone()
{
FileCopyPeerFA *c=new FileCopyPeerFA(session->Clone(),file,FAmode);
c->orig_url.set(orig_url);
return c;
}
// FileCopyPeerFDStream
#undef super
#define super FileCopyPeer
FileCopyPeerFDStream::FileCopyPeerFDStream(FDStream *o,dir_t m)
: FileCopyPeer(m), my_stream(o?o:new FDStream(1,"<stdout>")), stream(my_stream), close_when_done(o!=0)
{
Init();
}
FileCopyPeerFDStream::FileCopyPeerFDStream(const Ref<FDStream>& o,dir_t m)
: FileCopyPeer(m), stream(o), close_when_done(false)
{
Init();
}
void FileCopyPeerFDStream::Init()
{
seek_base=0;
create_fg_data=true;
need_seek=false;
can_seek = can_seek0 = stream->can_seek();
if(can_seek && stream->fd!=-1)
{
seek_base=lseek(stream->fd,0,SEEK_CUR);
if(seek_base==-1)
{
can_seek=false;
can_seek0=false;
seek_base=0;
}
}
if(stream->usesfd(1))
write_allowed=false;
if(mode==PUT)
put_ll_timer=new Timer(0,200);
if(mode==PUT && stream->fd==-1 && stream->can_setmtime())
stream->full_name.set(UseTempFile(stream->full_name));
}
void FileCopyPeerFDStream::Seek_LL()
{
int fd=stream->fd;
assert(fd!=-1);
if(CanSeek(seek_pos))
{
if(seek_pos==FILE_END)
{
seek_pos=lseek(fd,0,SEEK_END);
if(seek_pos==-1)
{
can_seek=false;
can_seek0=false;
seek_pos=0;
}
else
{
SetSize(seek_pos);
if(seek_pos>seek_base)
seek_pos-=seek_base;
else
seek_pos=0;
}
pos=seek_pos;
}
else
{
if(lseek(fd,seek_pos+seek_base,SEEK_SET)==-1)
{
can_seek=false;
can_seek0=false;
seek_pos=0;
}
pos=seek_pos;
}
if(mode==PUT)
pos+=Size();
}
else
{
seek_pos=pos;
}
}
int FileCopyPeerFDStream::getfd()
{
if(!stream)
return -1;
if(stream->fd!=-1)
return stream->fd;
int fd=stream->getfd();
if(fd==-1)
{
if(stream->error())
{
SetError(stream->error_text);
current->Timeout(0);
}
else
{
current->TimeoutS(1);
}
return -1;
}
stream->clear_status();
pos=0;
if(mode==PUT)
pos+=Size();
Seek_LL();
return fd;
}
int FileCopyPeerFDStream::Do()
{
int m=STALL;
if(Done() || Error())
return m;
if(do_mkdir) {
do_mkdir=false;
create_directories(dirname(stream->full_name).get_non_const());
}
if(verify)
{
if(verify->Error())
{
SetError(verify->ErrorText());
m=MOVED;
}
else if(verify->Done())
{
if(suggested_filename && stream && stream->full_name && auto_rename)
{
const char *new_name=dir_file(dirname(stream->full_name),suggested_filename);
struct stat st;
if(temp_file || (lstat(new_name,&st)==-1 && errno==ENOENT) || ResMgr::QueryBool("xfer:clobber",0)) {
debug((5,"copy: renaming `%s' to `%s'\n",stream->full_name.get(),suggested_filename.get()));
int res=rename(stream->full_name,new_name);
if(res==-1 && errno==EIO) {
// FUSE with HadoopFS workaround
unlink(new_name);
res=rename(stream->full_name,new_name);
}
if(res==-1) {
const char *err=xstring::format("rename(%s, %s): %s\n",stream->full_name.get(),new_name,strerror(errno));
if(temp_file)
SetError(err);
else
debug((3,"%s\n",err));
}
}
}
done=true;
m=MOVED;
}
return m;
}
bool check_min_size=true;
#ifndef NATIVE_CRLF
if(ascii)
check_min_size=false;
#endif
int res;
switch(mode)
{
case PUT:
if(Size()==0)
{
if(eof)
{
// make sure the stream is open - it may create an empty file.
if(stream && !stream->is_closed() && getfd()==-1)
return m;
if(!date_set && date!=NO_DATE && do_set_date)
{
if(date==NO_DATE_YET)
return m;
stream->setmtime(date);
date_set=true;
m=MOVED;
}
if(stream && close_when_done && !stream->Done())
return m;
if(!verify && do_verify)
verify=new FileVerificator(stream);
else
done=true;
return MOVED;
}
if(seek_pos==0)
return m;
}
if(!write_allowed)
return m;
if(getfd()==-1)
return m;
if(check_min_size && !eof && Size()<PUT_LL_MIN
&& put_ll_timer && !put_ll_timer->Stopped())
break;
res=Put_LL(buffer+buffer_ptr,Size());
if(res>0)
buffer_ptr+=res;
if(res!=0)
m=MOVED;
break;
case GET:
if(eof)
return m;
res=TuneGetSize(Get_LL(get_size));
if(res>0)
{
EmbraceNewData(res);
SaveMaxCheck(0);
}
if(res!=0 || eof)
m=MOVED;
break;
}
return m;
}
bool FileCopyPeerFDStream::IOReady()
{
return seek_pos==pos || stream->fd!=-1;
}
void FileCopyPeerFDStream::Seek(off_t new_pos)
{
if(pos==new_pos)
return;
#ifndef NATIVE_CRLF
if(ascii && new_pos!=0)
{
// it is possible to read file to determine right position,
// but it is costly.
can_seek=false;
// can_seek0 is still true.
return;
}
#endif
super::Seek(new_pos);
int fd=stream->fd;
if(fd==-1)
{
if(seek_pos!=FILE_END)
{
pos=seek_pos;
if(mode==PUT)
pos+=Size();
return;
}
else
{
off_t s=stream->get_size();
if(s!=-1)
{
SetSize(s);
pos=seek_pos+((mode==PUT)?Size():0);
return;
}
else
{
// ok, have to try getfd.
fd=getfd();
}
}
if(fd==-1)
return;
}
Seek_LL();
}
int FileCopyPeerFDStream::Get_LL(int len)
{
int res=0;
int fd=getfd();
if(fd==-1)
return 0;
if((want_date && date==NO_DATE_YET)
|| (want_size && size==NO_SIZE_YET))
{
struct stat st;
if(fstat(fd,&st)==-1)
{
SetDate(NO_DATE);
SetSize(NO_SIZE);
}
else
{
SetDate(st.st_mtime);
SetSize(S_ISREG(st.st_mode)?st.st_size:NO_SIZE);
#ifndef NATIVE_CRLF
if(ascii)
SetSize(NO_SIZE);
#endif
}
}
if(need_seek) // this does not combine with ascii.
lseek(fd,seek_base+pos,SEEK_SET);
char *p=GetSpace(ascii?len*2:len);
res=read(fd,p,len);
if(res==-1)
{
if(E_RETRY(errno))
{
Block(fd,POLLIN);
return 0;
}
if(stream->NonFatalError(errno))
return 0;
stream->MakeErrorText();
SetError(stream->error_text);
return -1;
}
stream->clear_status();
#ifndef NATIVE_CRLF
if(ascii)
{
for(int i=res; i>0; i--)
{
if(*p=='\n')
{
memmove(p+1,p,i);
*p++='\r';
res++;
}
p++;
}
}
#endif
if(res==0) {
debug((10,"copy-peer: EOF on FD %d\n",fd));
eof=true;
}
return res;
}
int FileCopyPeerFDStream::Put_LL(const char *buf,int len)
{
if(len==0)
return 0;
int fd=getfd();
if(fd==-1)
return 0;
int skip_cr=0;
#ifndef NATIVE_CRLF
if(ascii)
{
// find where line ends.
const char *cr=buf;
for(;;)
{
cr=(const char *)memchr(cr,'\r',len-(cr-buf));
if(!cr)
break;
if(cr-buf<len-1 && cr[1]=='\n')
{
skip_cr=1;
len=cr-buf;
break;
}
if(cr-buf==len-1)
{
if(eof)
break;
len--;
break;
}
cr++;
}
}
#endif // NATIVE_CRLF
if(len==0)
return skip_cr;
if(need_seek) // this does not combine with ascii.
lseek(fd,seek_base+pos-Size(),SEEK_SET);
int res=write(fd,buf,len);
if(res<0)
{
if(E_RETRY(errno))
{
Block(fd,POLLOUT);
return 0;
}
if(errno==EPIPE)
{
broken=true;
buffer.truncate(buffer_ptr);
eof=true;
return -1;
}
if(stream->NonFatalError(errno))
{
// in case of full disk, check file correctness.
if(errno==ENOSPC && can_seek)
{
struct stat st;
if(fstat(fd,&st)!=-1)
{
if(st.st_size<seek_base+pos-Size())
{
// workaround solaris nfs bug. It can lose data if disk is full.
if(buffer_ptr>=seek_base+pos-Size()-buffer_ptr-st.st_size)
UnSkip(seek_base+pos-Size()-st.st_size);
else
{
Empty();
pos=st.st_size;
}
}
}
}
return 0;
}
stream->MakeErrorText();
SetError(stream->error_text);
return -1;
}
stream->clear_status();
if(res==len && skip_cr)
{
res+=skip_cr;
// performance gets worse because of writing a single char,
// but leaving uncomplete line on screen allows mixing it with debug text.
if(write(fd,"\n",1)==1)
res+=1;
}
if(put_ll_timer)
put_ll_timer->Reset();
return res;
}
FgData *FileCopyPeerFDStream::GetFgData(bool fg)
{
if(!my_stream || !create_fg_data)
return 0; // if we don't own the stream, don't create FgData.
if(stream->GetProcGroup())
return new FgData(stream->GetProcGroup(),fg);
return 0;
}
void FileCopyPeerFDStream::WantSize()
{
struct stat st;
int res=-1;
if(stream->fd!=-1)
res=fstat(stream->fd,&st);
else if(stream->full_name)
res=stat(stream->full_name,&st);
if(res!=-1)
SetSize(S_ISREG(st.st_mode)?st.st_size:NO_SIZE);
else
super::WantSize();
}
void FileCopyPeerFDStream::RemoveFile()
{
stream->remove();
removing=false; // it is instant.
file_removed=true;
Suspend();
current->Timeout(0);
}
const char *FileCopyPeerFDStream::GetStatus()
{
if(verify)
return verify->Status();
return stream->status;
}
void FileCopyPeerFDStream::Kill(int sig)
{
stream->Kill(sig);
}
FileCopyPeerFDStream *FileCopyPeerFDStream::NewPut(const char *file,bool cont)
{
int flags=O_WRONLY|O_CREAT;
if(!cont) {
flags|=O_TRUNC;
if(!ResMgr::QueryBool("xfer:clobber",0))
flags|=O_EXCL;
}
return new FileCopyPeerFDStream(new FileStream(file,flags),
FileCopyPeer::PUT);
}
FileCopyPeerFDStream *FileCopyPeerFDStream::NewGet(const char *file)
{
return new FileCopyPeerFDStream(new FileStream(file,O_RDONLY),
FileCopyPeer::GET);
}
FileCopyPeer *FileCopyPeerFDStream::Clone()
{
NeedSeek();
FileCopyPeerFDStream *peer=new FileCopyPeerFDStream(stream,mode);
peer->NeedSeek();
peer->SetBase(0);
return peer;
}
// FileCopyPeerDirList
FileCopyPeerDirList::FileCopyPeerDirList(FA *s,ArgV *v)
: FileCopyPeer(GET), session(s)
{
dl=session->MakeDirList(v);
if(dl==0)
eof=true;
can_seek=false;
can_seek0=false;
}
int FileCopyPeerDirList::Do()
{
if(Done())
return STALL;
if(dl->Error())
{
SetError(dl->ErrorText());
return MOVED;
}
const char *b;
int s;
dl->Get(&b,&s);
if(b==0) // eof
{
eof=true;
return MOVED;
}
if(s==0)
return STALL;
memcpy(GetSpace(s),b,s);
SpaceAdd(s);
dl->Skip(s);
return MOVED;
}
// FileCopyPeerMemory
int FileCopyPeerMemory::Do()
{
int m=STALL;
if(mode==PUT) {
max_buf=max_size+1;
if(Size()>max_size) {
SetError("buffer limit exceeded");
broken=true;
return MOVED;
}
}
return m;
}
// FileVerificator
void FileVerificator::Init0()
{
done=false;
if(!ResMgr::QueryBool("xfer:verify",0)
|| ResMgr::Query("xfer:verify-command",0).is_empty())
done=true;
}
void FileVerificator::InitVerify(const char *f)
{
if(done)
return;
ArgV *args=new ArgV(ResMgr::Query("xfer:verify-command",0));
args->Append(f);
Log::global->Format(9,"running %s %s\n",args->a0(),f);
verify_process=new InputFilter(args);
verify_process->StderrToStdout();
verify_buffer=new IOBufferFDStream(verify_process.Cast<FDStream>(),IOBuffer::GET);
}
FileVerificator::FileVerificator(const char *f)
{
Init0();
InitVerify(f);
}
FileVerificator::FileVerificator(const FDStream *stream)
{
Init0();
if(done)
return;
const char *f=stream->full_name;
if(!f)
{
done=true;
return;
}
const char *cwd=stream->GetCwd();
int cwd_len=xstrlen(cwd);
if(cwd && cwd_len>0 && !strncmp(f,cwd,cwd_len))
{
f+=cwd_len;
while(*f=='/')
f++;
if(*f==0)
f=".";
}
InitVerify(f);
if(verify_process)
{
verify_process->SetProcGroup(stream->GetProcGroup());
verify_process->SetCwd(cwd);
}
}
FileVerificator::FileVerificator(const FileAccess *session,const char *f)
{
Init0();
if(done)
return;
if(strcmp(session->GetProto(),"file"))
{
done=true;
return;
}
InitVerify(f);
verify_process->SetCwd(session->GetCwd());
}
FileVerificator::~FileVerificator() {}
int FileVerificator::Do()
{
int m=STALL;
if(done)
return m;
verify_process->Kill(SIGCONT);
if(!verify_buffer->Eof())
return m;
if(verify_process->GetProcState()!=ProcWait::TERMINATED)
return m;
done=true;
m=MOVED;
if(verify_process->GetProcExitCode()!=0)
{
error_text.set(verify_buffer->Get());
error_text.rtrim('\n');
if(error_text.length()==0)
error_text.set(_("Verify command failed without a message"));
const char *nl=strrchr(error_text,'\n');
if(nl)
error_text.set(nl+1);
}
return m;
}
// special pointer to creator of ftp/ftp copier. It is init'ed in Ftp class.
FileCopy *(*FileCopy::fxp_create)(FileCopyPeer *src,FileCopyPeer *dst,bool cont);