bool DBManager::writeFinishedTask()
{
    TaskPtr pTask;
    finishedTaskMgr_->lock();
    if(finishedTaskMgr_->hasTask())
    {
        pTask = finishedTaskMgr_->popTask();
    }
    finishedTaskMgr_->unlock();

    if(!pTask)
    {
        return true;
    }

    if(pTask->input().ProblemID == 0)//IDE测试功能，不写数据库
    {
        return true;
    }
    
    if(!writeToDB(pTask))
    {
        return false;
    }

    OJCout<<GetOJString("write finished:")<<pTask->input().SolutionID<<std::endl;
    return true;
}

std::vector<TaskPtr> ThreadGroup::remove(TaskGenerator* generator)
{
    std::vector<TaskPtr> remaining_tasks;

    std::unique_lock<std::recursive_mutex> lock(tasks_mtx_);

    TaskGeneratorPtr removed;

    for (auto it = tasks_.begin(); it != tasks_.end();) {
        TaskPtr task = *it;
        if (task->getParent() == generator) {
            remaining_tasks.push_back(task);
            it = tasks_.erase(it);
        } else {
            ++it;
        }
    }

    for (auto it = generators_.begin(); it != generators_.end();) {
        if (it->get() == generator) {
            removed = *it;
            it = generators_.erase(it);
        } else {
            ++it;
        }
    }

    work_available_.notify_all();

    generator_connections_[generator].clear();

    generator_removed(removed);

    return remaining_tasks;
}

void Peer::HandleResponses(WMessagePtr pWMsg)
{
	MessagePtr pMsg = pWMsg->GetMsg();

	// get the peer list if given in message
	std::vector<RemotePeers> peerList = pMsg->m_peerList;
	if( peerList.size() > 0 ) {
		long ts = pMsg->m_timestamp;
		this->HandlePeerList(peerList, ts);
	}

	// see if it is a task message
	if( pMsg->IsTaskMessage() ) {
		int msgTaskId = pMsg->m_taskId;
		// get the task and forward the message
		std::map<int, TaskPtr>::iterator it;
		it = m_taskList.find(msgTaskId);
		if( it != m_taskList.end() ) {
			TaskPtr pTask = (*it).second;
			pTask->GetMessageQueue()->PutMessage(pWMsg);
		}
		else {
			Log(CONSOLE, L"No Task Related to Task Response Message Received for taskId: %d\n", msgTaskId);
		}
		return;
	}

	switch(pMsg->m_respType) {
		case GET_ONLINE_RESP:
			HandleOnlineResp(pMsg);
			break;
		default:
			break;
	}
}

void ThreadGroup::executeTask(const TaskPtr& task)
{
    try {
        std::unique_lock<std::recursive_mutex> state_lock(execution_mtx_);
        ProfilerPtr profiler = getProfiler();
        Trace::Ptr interlude;
        if (profiler && profiler->isEnabled()) {
            TimerPtr timer = profiler->getTimer(getName());
            interlude.reset(new Trace(timer, task->getName()));
        }

        task->execute();

    } catch (const std::exception& e) {
        TaskGenerator* gen = task->getParent();
        if (gen) {
            gen->setError(e.what());
        }
    } catch (const std::string& s) {
        std::cerr << "Uncaught exception (string) exception: " << s << std::endl;

    } catch (const csapex::Failure& assertion) {
        handler_.handleAssertionFailure(assertion);

    } catch (...) {
        std::cerr << "Uncaught exception of unknown type and origin in execution of task " << task->getName() << "!" << std::endl;
        throw;
    }
}

void* SchedulerProcessor() {
    bool flag = true;
    while(true) {
        vector<string> machines = MachinePoolI::Instance()->GetAllMachineEndpoint();
        if(machines.size() == 0) {
            //没有机器,不能一直查询
            //LOG4CPLUS_INFO(logger, "no machine right now...");
            sleep(1);
        } else { 
            if(flag) {
                LOG4CPLUS_INFO(logger, "begin to generate request");
                //有机器再启动线请求程,只能启动一次
                thread request_generator_t(RequestGenerator);
                //这个线程主要修改rate
                RequestI::Instance()->Start();
                flag = false;
            }
 
            for(vector<string>::iterator it = machines.begin(); it != machines.end(); it++) {
                TaskPtr task;
                TaskBufferI::Instance()->PopFront(&task);
                try {
                    Proxy<WorkerClient> proxy = Rpc<WorkerClient, WorkerClient>::GetProxy(*it);
                    proxy().StartTask(task->GetTaskInfo());
                } catch (TException &tx) {
                    LOG4CPLUS_WARN(logger, "machine " << *it << " is down");
                    MachinePoolI::Instance()->Delete(*it);
                }
            }
        } 
    }
    return NULL;
}

	void GL2TextureManager::makeLoadFullResource(ResourceMeta* meta, GL2Texture* texture)
	{
		texture->m_loaded = GL2Texture::TextureLoadState::LOADING;
		ResourceLoader::RawResourceTask rawDataTask = loadRawData(loader, meta);
		TaskPtr textureLoader = std::make_shared<RawTextureLoader>(this, texture, rawDataTask);
		textureLoader->addDependency(rawDataTask);
		loader->getEngine()->getTaskManager()->addTask(textureLoader);
	}

void VC::RemoveTask(const TaskPtr& task) {
    //里面有并发控制机制
    TaskState ts = task->GetTaskState();
    assert(ts == TASK_WAIT || ts == TASK_RUN);
    if(ts == TASK_WAIT) {
        m_wait_queue.Erase(task->GetId());    
    } else {
        m_run_queue.Erase(task->GetId());
    }
}

 void ThreadPool::PoolWorker()
 {
     TaskPtr task;
     while (running_)
     {
         task_event_.wait();
         while (running_ && task_queue_.try_pop(task))
         {
             task->Run();
         }
     }
 }

bool TaskViewImpl::updateTask(Task* task)
{
    bool updated = false;
    
    int index = taskCombo.findText(task->name().c_str());

    if(index < 0 || index >= static_cast<int>(tasks.size())){
        addTask(task);
    } else {
        if(isWaiting()){
            mv->putln(MessageView::WARNING,
                      format(_("Task \"%1%\" cannot be updated now because it is wating for a command to finish."))
                      % task->name());
        } else {
            TaskInfo& info = tasks[index];
            TaskPtr oldTask = info.task;
            info.task = task;

            bool doEmitSignals = task != oldTask;

            if(index != currentTaskIndex){
                if(doEmitSignals){
                    sigTaskRemoved(oldTask);
                    sigTaskAdded(task);
                }
            } else {
                info.state = new Mapping();

                if(isExecutionEnabled()){
                    oldTask->storeState(self, *info.state);
                }
                if(doEmitSignals){
                    sigTaskRemoved(oldTask);
                }
                
                setCurrentTask(index, true);

                if(doEmitSignals){
                    sigTaskAdded(task);
                }

                if(isExecutionEnabled()){
                    task->restoreState(self, *info.state);
                }
            }
            os << format(_("Task \"%1%\" has been updated with the new one.")) % task->name() << endl;
            updated = true;
        }
    }

    return updated;
}

	static void _exec(executor* exec) {
		while (true) {
			TaskPtr task = exec->pull();
			if (task) {
				task->run();
				continue;
			}
			if (!exec->_running) {
				std::cout << "Finished running all task " << exec << std::endl;
				return;
			}
		};
	}

SingleTaskMenu::SingleTaskMenu(const QString& tablename, CamcopsApp& app) :
    MenuWindow(app, ""),  // start with a blank title
    m_tablename(tablename),
    m_current_task(nullptr)
{
    // Title
    TaskFactoryPtr factory = app.factory();
    TaskPtr specimen = factory->create(m_tablename);
    m_title = specimen->menutitle();
    m_anonymous = specimen->isAnonymous();

    // m_items is EXPENSIVE (and depends on security), so leave it to build()
}

// handle the connection failures
void Peer::HandleConnectionFailure(WMessagePtr pMsg)
{
	// see to which peer-node the connection failed
	// see if there is any task been alloted to that peer
	// see if there is any job been executed for that peer

	int failedConnPeer = pMsg->GetPeerId();
	// get the task and forward the failure message
	std::map<int, TaskPtr>::iterator it = m_taskList.begin();
	for( ;it != m_taskList.end(); it++ ) {
		TaskPtr pTask = (*it).second;
		pTask->GetMessageQueue()->PutMessage(pMsg);
	}
}

int ThreadPool::threadRun(){
	LOG_INFO("ThreadPool " << name() << " workThread start");
	
	while (_started) {
		TaskPtr t = getNextTask();
		int ret = t->execute();
		if (ret < 0) {
			LOG_ERROR("ThreadPool " << name() << "task execution error");
		}
	}

	LOG_INFO("ThreadPool stoped");
	return 0;
}

void Server::HandleTasks()
{
  //logs::Debug("Handling listener tasks..");
  TaskPtr task;
  while (true)
  {
    {
      std::lock_guard<std::mutex> lock(queueMutex);
      if (queue.empty()) break;
      task = queue.front();
      queue.pop();
    }

    task->Execute(*this);
  }
}

/// choose a task and assign it to cellet
void* ScheduleProcessor(void* unused) {
    while (true) {
        // pick a task from the framework pool
        TaskPtr task = ComponentsMgr::Instance()->GetPool()->GetTask();
        // has wait task then begin to assign the task
        if (task.get()) {
            // task asssign success change task state to started
            if (task->AssignTask())
                task->TaskAssigned();
            else
                ComponentsMgr::Instance()->GetPool()->AddTask(task);
        }
        usleep(1000*50);
    }
    return NULL;
}

void SingleTaskMenu::build()
{
    TaskFactoryPtr factory = m_app.factory();
    TaskPtr specimen = factory->create(m_tablename);

    // Common items
    QString info_icon_filename = UiFunc::iconFilename(UiConst::ICON_INFO);
    m_items = {
        MenuItem(tr("Options")).setLabelOnly(),
        MAKE_CHANGE_PATIENT(app),
        MenuItem(
            tr("Task information"),
            HtmlMenuItem(
                m_title,
                FileFunc::taskHtmlFilename(specimen->infoFilenameStem()),
                info_icon_filename),
            info_icon_filename
        ),
        MenuItem(tr("Task instances") + ": " + m_title).setLabelOnly(),
    };

    // Task items
    TaskPtrList tasklist = factory->fetch(m_tablename);
    qDebug() << Q_FUNC_INFO << "-" << tasklist.size() << "tasks";
    for (auto task : tasklist) {
        m_items.append(MenuItem(task, false));
    }

    // Call parent buildMenu()
    MenuWindow::build();

    // Signals
    connect(&m_app, &CamcopsApp::selectedPatientChanged,
            this, &SingleTaskMenu::selectedPatientChanged,
            Qt::UniqueConnection);
    connect(&m_app, &CamcopsApp::taskAlterationFinished,
            this, &SingleTaskMenu::taskFinished,
            Qt::UniqueConnection);
    connect(this, &SingleTaskMenu::offerAdd,
            m_p_header, &MenuHeader::offerAdd,
            Qt::UniqueConnection);
    connect(m_p_header, &MenuHeader::addClicked,
            this, &SingleTaskMenu::addTask,
            Qt::UniqueConnection);

    emit offerAdd(m_anonymous || m_app.patientSelected());
}

//根据状态插入到相应队列中
void VC::PushTask(const TaskPtr& task) {
    TaskState ts = task->GetTaskState(); 
    assert(ts == TASK_WAIT || ts == TASK_RUN);
    if(ts == TASK_WAIT) 
        m_wait_queue.PushBack(task);
    else 
        m_run_queue.PushBack(task);
}

// release m_used_cpu and m_used_memory of group
bool GroupPool::ReleaseUsedResource(const TaskPtr& task_ptr) {
    GroupPtr group_ptr = GetGroupPtr(task_ptr->GetGroupName());
    if (!group_ptr) {
        LOG4CPLUS_ERROR(logger, "No this group, group name: " << task_ptr->GetGroupName());
        return false;
    }

    double factor = 1.0;
    int32_t priority = task_ptr->GetRawPriority();
    if (JOB_RAW_PRIO_PRIVILEGE == priority) {
        factor = m_privilege_factor;
    }

    group_ptr->SetUsedCPU(task_ptr->GetNeedCpu() * factor, true);
    group_ptr->SetUsedMemory(task_ptr->GetNeedMemory() * factor, true);
    return true;
}

void SingleTaskMenu::addTask()
{
    // The task we create here needs to stay in scope for the duration of the
    // editing! The simplest way is to use a member object to hold the pointer.
    TaskFactoryPtr factory = m_app.factory();
    TaskPtr task = factory->create(m_tablename);
    if (!task->isAnonymous()) {
        int patient_id = m_app.currentPatientId();
        if (patient_id == DbConst::NONEXISTENT_PK) {
            qCritical() << Q_FUNC_INFO << "- no patient selected";
            return;
        }
        task->setPatient(m_app.currentPatientId());
    }
    task->save();
    OpenableWidget* widget = task->editor();
    m_app.open(widget, task, true);
}

void Peer::HandleTaskResult(WMessagePtr pWMsg)
{
	MessagePtr pMsg = pWMsg->GetMsg();
	Log(CONSOLE, L"Handling Task Result from Peer: %d, taskid: %d\n", 
											pMsg->m_fromPeerId, pMsg->m_taskId);

	PeerNodePtr pNode = pMsg->m_conn->GetPeerNode();
	// get the task and forward the message
	std::map<int, TaskPtr>::iterator it;
	it = m_taskList.find(pMsg->m_taskId);
	if( it != m_taskList.end() ) {
		TaskPtr pTask = (*it).second;
		pTask->GetMessageQueue()->PutMessage(pWMsg);
	}
	else {
		Log(CONSOLE, L"No Task Related to Task Result Message Received for taskId: %d\n", pMsg->m_taskId);
	}
}