engine.cpp

Last change: 2008-02-03 This is the source code of KCeasy。

    OutQueueCritSec.Leave();

    LockSearches();
    while(!Searches.empty()) {
        delete Searches.front();
        Searches.pop_front();
    }
    ReleaseSearches();

    LockDownloads();
    while(!Downloads.empty()) {
        delete Downloads.front();
        Downloads.pop_front();
    }
    ReleaseDownloads();

    LockUploads();
    while(!Uploads.empty()) {
        delete Uploads.front();
        Uploads.pop_front();
    }
    ReleaseUploads();

    delete Shares; Shares = NULL;
    delete Networks; Networks = NULL;

    SetState(Offline);
    UsingLocalGift = false;
    NextGiftId = MinGiftId;
    GiftHost = "";
    GiftPort = 0;
    GiftLog = "";
}

void TEngine::SocketThreadFunc()
{
    TGiftCommand* Cmd;
    TStringSocket Socket;
    char* Data = new char[1024];
    int DataLen = 0;
    int DataSize = 1024;

    // connect to daemon
    if(!Socket.Connect(GiftHost.c_str(),GiftPort)) {
        if(!IsUsingLocalGift()) {
            SendCallback(CbcNetworkError,(void*)&Socket.GetErrorString());
            SetState(ConnectFailed, true);
            Reset();
            return;
        }
        SetState(LaunchingGift, true);

        // if there is a lingering daemon from a previous session kill it
        if(GiftLauncher->IsRunning())
            GiftLauncher->Kill();

        if(!GiftLauncher->Launch()) {
            SetState(LaunchFailed, true, true); // gift closed connection unexpectedly
            Reset();
            return;
        }
        SetState(LaunchSucceeded, true);
        SetState(Connecting, true);
        for(int i=0;!Socket.Connect(GiftHost.c_str(),GiftPort);++i) {
            if(i >= LOCAL_CONNECT_RETRY_COUNT) {
                // we started giFT but can't connect, kill it
                GiftLauncher->Kill();
                SendCallback(CbcNetworkError,(void*)&Socket.GetErrorString());
                SetState(LaunchFailed, true);
                Reset();
                return;
            }
            Sleep(LOCAL_CONNECT_RETRY_DELAY);
        }
    }

    Cmd = new TGiftCommand("ATTACH");
    Cmd->AddNode(TGiftCommand("client",ClientName.c_str()));
    Cmd->AddNode(TGiftCommand("version",ClientVersion.c_str()));
    QueueCommand(Cmd);

    unsigned int TimerFuncNextRaise = ::GetTickCount() + TIMER_FUNC_INTERVAL;
    AutoSearchCountdown = 0;

    while(!ExitSocketThread) {
        if(Socket.WaitForData(10)) {
            char* Buf;

            if((Buf = Socket.Receive()) == NULL)
                break;

/*
    FILE* fp = fopen("part_network.log","ab");
    fprintf(fp,"%s",Buf);
    fclose(fp);
*/
            int AddLen = strlen(Buf);

            // resize Data buffer if necessary
            if(DataSize < DataLen + AddLen + 1) {
                char* OldData = Data;
                while(DataSize < DataLen + AddLen + 1)
                    DataSize *= 2;
                Data = new char[DataSize];
                memcpy(Data, OldData, DataLen + 1);
                delete[] OldData;
            }

            // data data just read
            memcpy(Data + DataLen,Buf,AddLen+1);
            Buf = Data + DataLen;
            DataLen += AddLen;

            while(((Buf = strchr(Buf,';')) != NULL) && !ExitSocketThread) {
                if(Buf[-1] == '\\') {
                    Buf++;
                    continue;
                }

                Buf++; // move past ';'
                int PacketLen = Buf - Data;

#ifdef NETWORK_LOGGING
                FILE* fp = fopen("network.log","ab");
                fprintf(fp,"\n\n*** (PacketLen: %d) ***\n",PacketLen);
                fwrite(Data,PacketLen,1,fp);
                fclose(fp);
#endif
                Cmd = new TGiftCommand();
                Cmd->Parse(Data);

                if(Debug)
                    SendCallback(CbcDebug,(void*)1,(void*)&Cmd->Print());
                if(!IsDisconnecting())
                    DispatchCommand(Cmd);

                delete Cmd;

                // remove processed data
                DataLen -= PacketLen;
                memmove(Data,Buf,DataLen+1); // also copy \0
                Buf = Data;
            }
        }
        // anything to send?
        if(!OutQueue.empty()) {
            OutQueueCritSec.Enter();
            Cmd = OutQueue.front();
            OutQueue.pop();
            OutQueueCritSec.Leave();

            if(Debug)
                SendCallback(CbcDebug,(void*)0,(void*)&Cmd->Print());
#ifdef NETWORK_LOGGING
            FILE* fp = fopen("network.log","ab");
            fwrite(Cmd->Print().c_str(),Cmd->Print().length(),1,fp);
            fclose(fp);
#endif
            if(!Socket.Send(Cmd->Print().c_str())) {
                SendCallback(CbcNetworkError,(void*)&Socket.GetErrorString());
                break;
            }

            // End thread if this was a QUIT or DETACH.
            if(Cmd->Name == "QUIT" || Cmd->Name == "DETACH") {
                Socket.Shutdown();
                ExitSocketThread = true;
            }

            delete Cmd;
        }
        // run timer func?
        if(GetTickCount() > TimerFuncNextRaise) {
            TimerFunc();
            TimerFuncNextRaise = GetTickCount() + TIMER_FUNC_INTERVAL;
        }

        // check for log output from giFT
        if(UsingLocalGift)
            ReadGiftLog();
    }
    Socket.Disconnect();
    delete[] Data;

//    GetLauncher()->CloseStdoutHandle();

    DisconnectTime = ::GetTickCount() / 1000;

    if(ExitSocketThread)
        SetState(Offline, true);
    else
        SetState(Offline, true, true); // gift closed connection unexpectedly

    // kceasy may have started a reconnect... this is so messy
    if(IsOffline())
        Reset();
}

// called periodically every couple seconds by the engine thread
void TEngine::TimerFunc()
{
    // update network stats periodically
    Networks->UpdateIfNecessary();

    // keep downloads alive
    if (AutoSearchCountdown > 0)
        AutoSearchCountdown--;
    LockDownloads();

    TDownloadIterator itr = Downloads.begin();
    while(itr != Downloads.end() && AutoSearchCountdown <= 0 &&
          GetActiveHashSearchesCount() < MAX_AUTO_SEARCHES)
    {
        // oh well...
//        ReleaseDownloads();
        if((*itr)->KeepAlive())
            AutoSearchCountdown = AUTO_SEARCH_DELAY; // wait x seconds before next search
//        LockDownloads();
        ++itr;
    }
    ReleaseDownloads();

    // commit search results to UI
    LockSearches();
    for(TSearchIterator itr = Searches.begin(); itr != Searches.end(); ++itr) {
//        ReleaseSearches();
        if((*itr)->GetState() == TSearch::Searching)
            (*itr)->CommitCachedResults();
//        LockSearches();
    }
    ReleaseSearches();
}

bool TEngine::ReadGiftLog()
{
    char buf[2048];
    DWORD BytesAvail;
    DWORD BytesRead;
    int pos;

    // check if there is new data
    if(!PeekNamedPipe(GiftLauncher->GetStdoutHandle(),NULL,0,NULL,&BytesAvail,NULL))
        return false;

    if(BytesAvail == 0)
        return false;

    if(BytesAvail > 2047)
        BytesAvail = 2047;

    // read new data
    if(!ReadFile(GiftLauncher->GetStdoutHandle(),buf,BytesAvail,&BytesRead,NULL))
        return false;

    if(BytesRead == 0)
        return false;

    // note: we must read from the pipe even if we don't need the data or giFT will block
    if(!GiftLogging)
        return false;

    // add new data to string
    buf[BytesRead] = 0;
    GiftLog += buf;

    // if all entire lines to application
    if((pos = GiftLog.rfind('\n')) == -1)
        return false;

    string Rest = GiftLog.substr(pos+1);
    GiftLog.resize(pos+1);

    SendCallback(CbcGiftLog,(void*)&GiftLog,NULL);

    GiftLog = Rest;
    return true;
}


void TEngine::QueueCommand(TGiftCommand* Cmd)
{
    OutQueueCritSec.Enter();
    OutQueue.push(Cmd);
    OutQueueCritSec.Leave();
}

void TEngine::DispatchCommand(TGiftCommand* Cmd)
{
    if(Cmd->Name == "ATTACH") {
        ConnectTime = ::GetTickCount() / 1000;
        SetState(Online, true);
        Networks->Update();
        return;
    } else if(Cmd->Name == "STATS") {
        Networks->ProcessStats(Cmd);
        return;
    } else if(Cmd->Name == "ITEM") {
        unsigned int GiftId = string_to_int(Cmd->Value);

        // is this a shares update?
        if(GiftId == Shares->FilesGiftId) {
            Shares->ProcessItem(Cmd);
            return;
        }
        // not a shares update, find corresponding search
        LockSearches();
        TSearchIterator sitr = Searches.begin();
        while(sitr != Searches.end() && (*sitr)->GiftId != GiftId)
            ++sitr;
        ReleaseSearches();

        if(sitr != Searches.end() && (*sitr)->State == TSearch::Searching)
            (*sitr)->ProcessItem(Cmd);

        return;
    } else if(Cmd->Name == "ADDDOWNLOAD") {
        // find corresponding download
        THashSet* Hashes = new THashSet();
        Hashes->AddGiftHash(Cmd->GetNode("hash").Value);
        LockDownloads();
        TDownloadIterator ditr = Downloads.begin();
        while(ditr != Downloads.end() && (*(*ditr)->GetHashes() != *Hashes || (*ditr)->State != TDownload::Registering))
            ++ditr;
        bool NewDownload = (ditr == Downloads.end());
        // new download after ATTACH
        if(NewDownload)
            ditr = Downloads.insert(Downloads.end(),new TDownload(this, Hashes));
        ReleaseDownloads();
        (*ditr)->ProcessAddDownload(Cmd);
        delete Hashes;

        return;
    } else if(Cmd->Name == "DELDOWNLOAD") {
        // find corresponding download
        unsigned int GiftId = string_to_int(Cmd->Value);
        LockDownloads();
        TDownloadIterator ditr = Downloads.begin();
        while(ditr != Downloads.end() && (*ditr)->GiftId != GiftId)
            ++ditr;
        ReleaseDownloads();

        if(ditr != Downloads.end())
            (*ditr)->ProcessDelDownload(Cmd);

        return;
    } else if(Cmd->Name == "CHGDOWNLOAD") {
        // find corresponding download
        unsigned int GiftId = string_to_int(Cmd->Value);
        LockDownloads();
        TDownloadIterator ditr = Downloads.begin();
        while(ditr != Downloads.end() && (*ditr)->GiftId != GiftId)
            ++ditr;
        ReleaseDownloads();

        if(ditr != Downloads.end())
            (*ditr)->ProcessChgDownload(Cmd);

        return;
    } else if(Cmd->Name == "ADDSOURCE") {
        // find download
        unsigned int GiftId = string_to_int(Cmd->Value);
        LockDownloads();
        TDownloadIterator ditr = Downloads.begin();
        while(ditr != Downloads.end() && (*ditr)->GiftId != GiftId)
            ++ditr;
         ReleaseDownloads();

        if(ditr != Downloads.end())
            (*ditr)->ProcessAddSource(Cmd);

        return;
    } else if(Cmd->Name == "DELSOURCE") {
        // find download
        unsigned int GiftId = string_to_int(Cmd->Value);
        LockDownloads();
        TDownloadIterator ditr = Downloads.begin();
        while(ditr != Downloads.end() && (*ditr)->GiftId != GiftId)
            ++ditr;
        ReleaseDownloads();

        if(ditr != Downloads.end())
            (*ditr)->ProcessDelSource(Cmd);

        return;
    } else if(Cmd->Name == "ADDUPLOAD") {
        // new download
        LockUploads();
        TUpload* Upload = new TUpload(this);
        Uploads.push_back(Upload);
        ReleaseUploads();

        Upload->ProcessAddUpload(Cmd);

        return;
    } else if(Cmd->Name == "DELUPLOAD") {
        // find corresponding upload
        unsigned int GiftId = string_to_int(Cmd->Value);
        LockUploads();
        TUploadIterator uitr = Uploads.begin();
        while(uitr != Uploads.end() && (*uitr)->GiftId != GiftId)
            ++uitr;
        ReleaseUploads();

        if(uitr != Uploads.end())
            (*uitr)->ProcessDelUpload(Cmd);

        return;
    } else if(Cmd->Name == "CHGUPLOAD") {
        // find corresponding upload
        unsigned int GiftId = string_to_int(Cmd->Value);
        LockUploads();
        TUploadIterator uitr = Uploads.begin();
        while(uitr != Uploads.end() && (*uitr)->GiftId != GiftId)
            ++uitr;
        ReleaseUploads();

        if(uitr != Uploads.end())
            (*uitr)->ProcessChgUpload(Cmd);

        return;
    } else if(Cmd->Name == "SHARE") {
        Shares->ProcessShare(Cmd);
        return;
    }
}

void TEngine::SetState(TState NState, bool NotifyUI, bool GiftCrashed)
{
    State = NState;
    if(NotifyUI)
        SendCallback(CbcStateChange, (void*)State, (void*)((int)GiftCrashed));
}

bool TEngine::SendCallback(TCallbackCode Code, void* Param1, void* Param2)
{
    if(!CallbackWindow)
        return false;
    TCallbackInfo CbInfo(Code,Param1,Param2);

#ifdef CALLBACK_LOGGING
    FILE* fp = fopen("callback.log","ab");
    fprintf(fp,"SendCallback(type = \"%d\", Param1 = 0x%08X, Param2 = 0x%08X)\r\n\tConnectState = %d\r\n",(int)Code,Param1,Param2,(int)ConnectState);
    fclose(fp);
#endif

    // using SendMessage makes windows switch to CallbackWindow's thread
    // and block this thread till the handler has executed
    SendMessage(CallbackWindow,WM_ENGINE_CALLBACK,(WPARAM)&CbInfo,(LPARAM)NULL);

#ifdef CALLBACK_LOGGING
    fp = fopen("callback.log","ab");
    fprintf(fp,"\treturned\r\n");
    fclose(fp);
#endif

    return true;
}

unsigned int TEngine::GetNextGiftId()
{
    NextGiftId++;

    if(NextGiftId >= MaxGiftId)
        NextGiftId = MinGiftId;

    return NextGiftId;
}

} // namespace KCeasyEngine