task.cpp

dget是一个基于Linux平台的多线程下载工具, 采用C++开发。主要支持FTP, HTTP, MMS, RTSP协议下载, 目前为止主要实现了FTP, HTTP, MMS,RTSP协议的多线程下载

/*
 * by balancesli
 * [e-mail]
 *
 * 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 2 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, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 */

#include <stdio.h>
#include <stdarg.h>
#include <string.h>
#include <stdlib.h>
#include <stdint.h>

#include <sys/stat.h>
#include <sys/time.h>
#include <sys/param.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <time.h>

#include <unistd.h>
#include <errno.h>
#include <assert.h>
#include <pthread.h>

//#include "Task.h"
#include "Utils.h"
#include "Url.h"
#include "FtpCli.h"
#include "HttpCli.h"
#include "dget.h"

#ifdef USE_QT
#include "listviews.h"
#include <qlistview.h>
#endif



//protect Thread Status after Thread Exec
//pthread_mutex_t StatusMutex = PTHREAD_MUTEX_INITIALIZER;
//pthread_cond_t ConnectingCond = PTHREAD_COND_INITIALIZER;

const char * ThreadStatusList[] = {
		"IDLE",
   		"CONNECTING",
   		"LOGININ",
   		"DOWNLOADING",
   		"COMPLETED",
   		"LOGINDENIED",
   		"CONNREFUSED",
   		"REMOTEFATAL",
   		"LOCALFATAL",
   		"TIMEOUT",
   		"MAXATTEMPTS"
};

void HttpThread(TThread * This)
{
    uerr_t err;
	
	THttpCli  hc(This->u);
	
   	gettimeofday(&This->BeginTime, NULL);

    do
    {

		if (This->TryAttempts > 0 && This->u->ResumeSupport == true)
		{
	    	if (This->Task->ResumeModifyHttpSingleConn(This) != 0)
	    	{
				This->Status = LOCALFATAL;
				return;
	    	}
	    	This->FileMode = strdup("ab");
		}

		This->TryAttempts++;

		if(This->Task->GetThreadCnt() == 1 && This->RemoteStartPos <= 0 && 
											This->u->ResumeSupport == false)
	    	err = hc.GetCompleteFile(This);
		else
	    	err = hc.GetFileChunk(This);

		if (err == HOK)
		{
			ShowMsg("%s have finished its task", This->Name);
	    	return;
		}

		switch (err)
		{
		case FWRITEERR:
	    	ShowMsg("Error writing to  file %s : %s", 
											This->LocalFile, strerror(errno));
	    	return;
	    	break;
		case FOPENERR:
	    	ShowMsg("Error opening file %s for writing: %s",
		    								This->LocalFile, strerror(errno));
	    	return;
	    	break;

		case CONERROR:
			if (errno == ETIMEDOUT)
				ShowMsg("Connection Attempt Timed out ...");
			break;

		case CONREJECT:
	    	ShowMsg("Connection attempt rejected ...");
	    	break;

		default:
	    	ShowMsg("Error occured in connection ...");
	    	break;

		}
    }while (This->TryAttempts < MAXATTEMPTS);

    This->Status = MAXTRYS;
	pthread_exit(0);
}

void FtpThread(TThread * This)
{
   	uerr_t err;

	TFtpCli  fc(This->u);
   	
	gettimeofday(&This->BeginTime, NULL);
	do
   	{
		if((This->TryAttempts > 0) && (This->u->ResumeSupport == true))
		{
		 	if(This->Task->ResumeModifyFtpSingleConn(This) != 0)
		    {
		    	This->Status = LOCALFATAL;
				return;
			}
		    This->FileMode = strdup("ab");
		}
		
		This->TryAttempts++;
		
		if (This->Task->GetThreadCnt() == 1)
	  		err = fc.GetFileToEnd(This);
		else
	  		err = fc.GetFileChunk(This);

		if (err == FTPOK)
		{	
			ShowMsg("%s has finished its task", This->Name);
			pthread_exit(0);
			return;
		}

		switch (err)
		{
		case FWRITEERR:
	    	ShowMsg("Error writing to  file %s : %s", 
								This->u->HostFileName, strerror(errno));
			pthread_exit(0);
	    	return;
	    	break;
			
		case FOPENERR:
	    	ShowMsg("Error opening file %s for writing: %s", 
									This->LocalFile, strerror(errno));
			pthread_exit(0);
	    	return;
	   		break;

		case CONERROR:
          if (errno == ETIMEDOUT)
			 ShowMsg("Connection Attempt Timed out ...");
		  else
		     ShowMsg("Error while connecting ...");
	   	  break;
		  
		case FTPLOGREFUSED:
		case FTPCONREFUSED:
		  	pthread_exit(0);
	    	return;
			break;
		default:
	    	ShowMsg("Error occured in connection ...");
	    	break;
		}

   	}while (This->TryAttempts < MAXATTEMPTS);
   	
	This->Status = MAXTRYS;

	pthread_exit(0);

}

/*********************TThread Impl*****************************/

TThread :: TThread(void)
{
	
	memset(Name, 0, 10);
	
	u = NULL;

	Status = UNKNOWN;
	
	LocalFile = NULL;
	FileMode = NULL;
	
	RemoteStartPos = 0;
	RemoteEndPos = 0;
	RemoteBytesReceived = 0;
	
	LocalStartPos = 0;
	
	Retry = true;
	TryAttempts = 0;
	
}

TThread :: ~TThread( void )
{

}

bool TThread :: Init( void )
{
    return false;
}


int TThread :: Exit( void )
{
    return 0;
}

void TThread :: Run( void )
{
		
}

TThread* TThread :: Self( void )
{
	
}



/********************* Task Impl*****************************/

TTask :: TTask(int n) : ThreadCnt(n)
{
	int i;

	if(n <= 0) 
		n = 2;

	if(n > MAXTHREADS) 
		n = MAXTHREADS;
	
	for(i = 0; i < n; i++)
    	ThreadQue[i] = new TThread();
		
	plog = NULL;
	u = NULL;
	
	memset(Name, 0, strlen(Name));
	
	Mode = NORMAL;
 	pthread_mutex_init(&StatusMutex, NULL);
    pthread_cond_init(&ConnectingCond, NULL);
}

TTask :: ~TTask(void)
{
	int i;
	
	for(i = 0; i < ThreadCnt; i++)
	{
		if(ThreadQue[i] != NULL)
			delete ThreadQue[i];
	}
	
	if( plog != NULL)
	{
		delete plog;	
		plog = NULL;
	}
	pthread_cond_destroy( &ConnectingCond ); 
	pthread_mutex_destroy( &StatusMutex );
}

/**********************************************/
// This ==> Current Thread 
int TTask :: CreateThread(TThread * This, ThreadFunc fn)
{
	int Retval = pthread_create(&This->ThreadId, NULL, fn, This);
	pthread_detach(This->ThreadId);
	return Retval;
}

void TTask :: JoinThread(TThread * This)
{
	pthread_join(This->ThreadId, NULL);
}

void TTask :: CancelThread(TThread * This)
{
	pthread_cancel(This->ThreadId);
}

void TTask :: ExitThread(TThread * This)
{
	pthread_exit(&This->Retval);
}

void TTask :: SupendThread(TThread * This)
{

}

void TTask :: ResumeThread(TThread * This)
{

}

void TTask :: LockThread(TTask * This) 
{
    pthread_mutex_lock(&This->StatusMutex); 
}

void TTask :: UnlockThread(TTask * This) 
{
    pthread_mutex_unlock(&This->StatusMutex);   
}

void TTask :: ThreadWaitCond(TTask * This) 
{
    pthread_cond_wait(&This->ConnectingCond, &This->StatusMutex);
}

void TTask :: BroadcastCond(TTask * This) 
{
    pthread_cond_broadcast(&This->ConnectingCond);
}

/*******************************************************/

float TTask :: GetCurThreadAvgSpeed(TThread * This)
{
	struct timeval CurTime;
	long TimeOfCost;
	float AvgSpeed;
	
	gettimeofday(&CurTime, NULL);
	TimeOfCost = CurTime.tv_sec - This->BeginTime.tv_sec;

	if(TimeOfCost > 0)
		AvgSpeed = (This->RemoteBytesReceived - This->LocalStartPos) / TimeOfCost;
    else    
		AvgSpeed = This->RemoteBytesReceived - This->LocalStartPos;

	return AvgSpeed / 1024;
}

float TTask :: GetCurThreadBytesReceived(TThread * This)
{
	return (float)This->RemoteBytesReceived / 1024;
}

float TTask :: GetCurThreadBlkSize(TThread * This)
{
	return (float)This->BlkSize / 1024;
}

const char * TTask :: GetCurThreadStatus(TThread * This)
{
	return ThreadStatusList[This->Status];
}

float TTask :: GetCurThreadPercent(TThread * This)
{
	 return (float)This->RemoteBytesReceived / This->BlkSize * 100;
}

long TTask :: GetAllRemoteBytesReceived(void)
{
	long AllRemoteBytesReceived = 0;
	int i;	
	for(i = 0; i < ThreadCnt; i++)
	{
		if(ThreadQue[i] != NULL)
            AllRemoteBytesReceived += ThreadQue[i]->RemoteBytesReceived;
    }
	
	return AllRemoteBytesReceived;
}

long TTask :: GetAllLocalBytesReceived(void)
{
	long AllLocalBytesReceived = 0;
	int i;	
	for(i = 0; i < ThreadCnt; i++)
	{
		if(ThreadQue[i] != NULL)
            AllLocalBytesReceived += ThreadQue[i]->LocalStartPos;
    }

	return AllLocalBytesReceived;
}

float TTask :: GetTotalAvgSpeed(void)
{
	long AllRemoteBytesReceived = 0;
	long AllLocalBytesReceived = 0;

	float TotalAvgSpeed = 0;
	struct timeval CurTime;
	long TimeOfCost;

    AllRemoteBytesReceived = GetAllRemoteBytesReceived();
    AllLocalBytesReceived = GetAllLocalBytesReceived();
	
	gettimeofday(&CurTime, NULL);

    TimeOfCost = CurTime.tv_sec - dlStartTime.tv_sec;

  	if(TimeOfCost > 0)
    	TotalAvgSpeed = (AllRemoteBytesReceived - AllLocalBytesReceived) / TimeOfCost;
	else
	    TotalAvgSpeed = AllRemoteBytesReceived - AllLocalBytesReceived;

	return TotalAvgSpeed;
}

void TTask :: SetdlStartTime(void)
{
	gettimeofday(&dlStartTime, NULL);
}

void TTask :: SetdlEndTime(void)
{
	gettimeofday(&dlEndTime, NULL);
}

long TTask :: GetTotalCostTime(void)
{
	long TimeOfCost = dlEndTime.tv_sec - dlStartTime.tv_sec;
	return TimeOfCost;
}

bool TTask :: IsAllCompleted(void)
{
	int i;
   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status != ALLDONE)
    		return false;
   	}
   	return true;
}

bool TTask :: IsAllLoginFailed(void)
{
   	int i;
   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status != LOGINFAIL)
    		return false;
   	}
   	return true;
}

bool TTask :: IsAllConnRejected(void)
{
   	int i;
   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status != CONREJECT)
    		return false;
   	}
   	return true;
}

bool TTask :: IsAllRemoteFailed(void)
{
   	int i;
   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status != REMOTEFATAL)
    		return false;
   	}
   	return true;
}

int TTask :: QueryAllDoneCnt(void)
{
	int i;
   	int Cnt = 0;

   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status == ALLDONE)
    		Cnt++;
   	}
   	return Cnt;
}

int TTask :: GetThreadCnt(void) const
{
	return ThreadCnt;
}

int TTask :: QueryConnectingCnt(void)
{
   	int i;
   	int Cnt = 0;

   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status == CONNECTING)
    		Cnt++;
   	}
   	return Cnt;
}

int TTask :: QueryLoggingCnt(void)
{
   	int i;
   	int Cnt = 0;

   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status == LOGGININ)
    		Cnt++;
   	}
   	return Cnt;
}

int TTask :: QueryDownloadingCnt(void)
{
   	int i;
   	int Cnt = 0;

   	for (i = 0; i < ThreadCnt; i++)
   	{
		if (ThreadQue[i]->Status == DOWNLOADING)
    		Cnt++;
   	}
   	return Cnt;
}

int TTask :: QueryLoginfailCnt(void)
{
   	int i;
   	int Cnt = 0;