void CPU_WaitStatus(condition_variable &cond, bool (*pred)()) {
	lock_guard guard(cpuThreadLock);
	while (!pred()) {
		cond.wait(cpuThreadLock);
	}
}

int main(int argc, char *argv[])
{
    if (argc != 4 && argc != 5)
    {
        cout << "Put \"[host-ipaddress:port] [authprovider] [authcode]\" for sign-up and sign-in"
             << endl;
        cout << "Put \"[host-ipaddress:port] [uid] [accessToken] 1\" for sign-in" <<
             endl;
        return 0;
    }

    OCPersistentStorage ps{ client_open, fread, fwrite, fclose, unlink };

    PlatformConfig cfg
    {
        ServiceType::InProc,
        ModeType::Both,
        "0.0.0.0", // By setting to "0.0.0.0", it binds to all available interfaces
        0,         // Uses randomly available port
        QualityOfService::LowQos,
        &ps
    };

    OCPlatform::Configure(cfg);

    OCStackResult result = OC_STACK_ERROR;

    string host = "coap+tcp://";
    host += argv[1];

    OCAccountManager::Ptr accountMgr = OCPlatform::constructAccountManagerObject(host,
                                       CT_ADAPTER_TCP);

    mutex blocker;
    unique_lock<mutex> lock(blocker);

    if (argc == 5)
    {
        accountMgr->signIn(argv[2], argv[3], &handleLoginoutCB);
        g_callbackLock.wait(lock);
    }
    else
    {
        accountMgr->signUp(argv[2], argv[3], &handleLoginoutCB);
        g_callbackLock.wait(lock);
        accountMgr->signIn(g_uid, g_accesstoken, &handleLoginoutCB);
        g_callbackLock.wait(lock);
    }

    // MQ broker resource
    g_mqBrokerResource = OCPlatform::constructResourceObject(host, DEFAULT_MQ_BROKER_URI,
                         static_cast<OCConnectivityType>(CT_ADAPTER_TCP | CT_IP_USE_V4), false,
    { string("oic.wk.ps") }, { string(DEFAULT_INTERFACE) });

    cout << "===Message Queue publisher sample===" << endl;
    cout << "PUT 0 to discover all topics" << endl;
    cout << "PUT 1 to discover type based topics" << endl;
    cout << "PUT 2 to select topic index for publishing data" << endl;
    cout << "PUT 3 to publish data to selected topic" << endl;
    cout << "PUT 4 to create topic" << endl;
    cout << "PUT 5 to create type based topic" << endl;

    string cmd;

    while (true)
    {
        cin >> cmd;

        try
        {

            QueryParamsMap query;
            OCRepresentation rep;
            string      topicType;

            switch (cmd[0])
            {
                case '0':
                    gTopicList.clear();
                    cout << "Discovering topics" << endl;
                    result = g_mqBrokerResource->discoveryMQTopics(query, &discoverTopicCB, QualityOfService::LowQos);
                    break;

                case '1':
                    gTopicList.clear();
                    cout << "Put topic type to discover: ";
                    cin >> cmd;
                    query["rt"] = cmd;
                    result = g_mqBrokerResource->discoveryMQTopics(query, &discoverTopicCB, QualityOfService::LowQos);
                    break;

                case '2':
                    cout << "Put discovered topic index to select: ";
                    cin >> cmd;
                    {
                        int index = atoi(cmd.c_str());
                        if (index < 0 || (unsigned int) index >= gTopicList.size())
                        {
                            cout << "invalid topic index selected" << endl;
                            continue;
//.........这里部分代码省略.........

    // running in a dedicated thread
    void process() {
        int clear_index;
        {
            unique_lock<mutex> auto_lock(cur_lock);
            cond.wait(auto_lock);
            clear_index = (cur_index + 2) % 3;
            int next_index = (cur_index + 1) % 3;
            cur_index = next_index;
            cur_set = &queue[next_index];
        }
        auto back_set = &queue[clear_index];

        auto it = back_set->begin();
        while (it != back_set->end()) {
            file_entry_t fe = *it;
            grandet_sync_task_t sync_task = NULL;
            grandet_remove_task_t remove_task = NULL;
            _files_lock.lock();
            INFO("processing %s\n", fe->link.c_str());
            if (fe->deleted) {
                INFO("processing deleted fe, size=%ld, fetched=%d, open_rc=%d\n", fe->st.st_size, fe->fetched, fe->open_rc);
            }
            if (fe->deleted && fe->open_rc == 0) {
                // when it is out of tree, no one can open it anymore 
                if (fe->deleted == 1) {
                    string data_path;
                    if (fe->fetched)
                        cache_remove(fe->st.st_size);
                    get_data_path(data_path, _data_store_base, fe->link.c_str());
                    grandet_remove_task_start(fe->link.c_str(), data_path.c_str(), &remove_task);
                    grandet_remove_task_wait(remove_task);
                    grandet_remove_task_end(remove_task);
                    
                    fe->deleted = 2; // avoiding repeated deletion
                }
                fe_put(fe);
            } else {
                // Update
                if (fe->content_dirty) {
                    if (!S_ISREG(fe->st.st_mode)) {
                        DEBUG("flush a non-regular file");
                    }
                    
                    string data_path;
                    get_data_path(data_path, _data_store_base, fe->link.c_str());
                    
                    fe->content_dirty = 0;
                    grandet_sync_task_start(fe->link.c_str(), data_path.c_str(), fe->xattr, &sync_task);
                    
                    _files_lock.unlock();
                    grandet_sync_task_wait(sync_task);
                    grandet_sync_task_end(sync_task);
                    _files_lock.lock();

                    // during the unlock and lock, there may be some
                    // thread writing data in and close, causing
                    // content_dirty be 1 again. Or some one can even
                    // delete it.  In those cases, it will appear in
                    // update queue again.
                    if (fe->open_rc == 0 && fe->fetched == 1 &&
                        fe->content_dirty == 0 && fe->deleted == 0) {
                        // DEBUG("add %s to idle list\n", fe->link.c_str());
                        // LRU
                        if (!list_empty(&fe->u_node))
                            list_del_init(&fe->u_node);
                        INFO("add %s to idle_list (dirty)\n", fe->link.c_str());
                        list_add_before(&_idle_list, &fe->u_node);
                    }
                } else if (fe->fetched == 1 && fe->open_rc == 0) {
                    if (!list_empty(&fe->u_node))
                        list_del_init(&fe->u_node);
                    INFO("add %s to idle_list (clean)\n", fe->link.c_str());
                    list_add_before(&_idle_list, &fe->u_node);
                }
                fe_put(fe);
            }
            _files_lock.unlock();
            ++ it;
        }
        back_set->clear();
    }

void CPU_SetState(CPUThreadState to) {
	lock_guard guard(cpuThreadLock);
	cpuThreadState = to;
	cpuThreadCond.notify_one();
	cpuThreadReplyCond.notify_one();
}

	void draw() {
		static int frameIndex = 0;
		ovrHmd_BeginFrame(hmd, frameIndex++);
		short textureSwitch = 0;// frameIndex % 2;

		// wait for camera thread
		{
			unique_lock<mutex> lck(camMutex);

			while (!camFinish){
				camFinishCondition.wait(lck);
			}
		}

		// swap image pointers
		for (int i = 0; i < 2; ++i)
		{
			IplImage* tempImage = perEyeReadImage[i];
			perEyeReadImage[i] = perEyeWriteImage[i];
			perEyeWriteImage[i] = tempImage;
		}

		// notify camera thread to resume
		{
			unique_lock<mutex> lck(camMutex);
			camFinish = false;
			drawFinish = true;
			drawFinishCondition.notify_all();
		}

		for (int i = 0; i < 2; ++i)
		{
			ovrEyeType eye = hmdDesc.EyeRenderOrder[i];

			if (perEyeReadImage[eye] && !terminateApp) {
				EyeArgs & eyeArgs = perEyeArgs[eye];
				gl::Stacks::projection().top() = eyeArgs.projection;
				gl::MatrixStack & mv = gl::Stacks::modelview();

				eyeArgs.framebuffer.activate();
				ovrPosef renderPose = ovrHmd_BeginEyeRender(hmd, eye);
				mv.withPush([&]{
					glClear(GL_DEPTH_BUFFER_BIT);
					//GlUtils::renderFloorGrid(glm::mat4());
					//GlUtils::renderSkybox(Resource::IMAGES_SKY_CITY_XNEG_PNG);

					if (ovrEye_Left == eye) {
						mv.rotate(M_PI / 2, glm::vec3(0, 0, 1));
					}
					else {
						mv.rotate(-M_PI / 2, glm::vec3(0, 0, 1));
					}

					mv.scale(1);
					mv.rotate(M_PI, glm::vec3(0, 1, 0));

					// only load necessary part of image
					glPixelStorei(GL_UNPACK_ROW_LENGTH, CAM_IMAGE_WIDTH);
					glPixelStorei(GL_UNPACK_SKIP_PIXELS, 189);

					// bind and load camera image
					imageTextures[eye]->bind();
					glTexSubImage2D(GL_TEXTURE_2D, 0, 0, 0, RENDER_IMAGE_WIDTH, RENDER_IMAGE_HEIGHT, GL_BGR, GL_UNSIGNED_BYTE, perEyeReadImage[eye]->imageData);

					glPixelStorei(GL_UNPACK_ROW_LENGTH, 0);
					glPixelStorei(GL_UNPACK_SKIP_PIXELS, 0);

					GlUtils::renderGeometry(quadGeom, texturedPtr);
				});

				ovrHmd_EndEyeRender(hmd, eye, renderPose, &perEyeArgs[eye].textures.Texture);
			}
			//eyeArgs.framebuffer.deactivate();
		};
		ovrHmd_EndFrame(hmd);
	}

	void release(){
		unique_lock<mutex> lck(mtx);
        ++count;
        cv.notify_all();
	}

	void updateCameraImages() {
		CvCapture* camLeft = cvCaptureFromCAM(701);
		cvSetCaptureProperty(camLeft, CV_CAP_PROP_FOURCC, CV_FOURCC('M', 'J', 'P', 'G'));
		cvSetCaptureProperty(camLeft, CV_CAP_PROP_FRAME_WIDTH, CAM_IMAGE_WIDTH);
		cvSetCaptureProperty(camLeft, CV_CAP_PROP_FRAME_HEIGHT, CAM_IMAGE_HEIGHT);

		CvCapture* camRight = cvCaptureFromCAM(700);
		cvSetCaptureProperty(camRight, CV_CAP_PROP_FOURCC, CV_FOURCC('M', 'J', 'P', 'G'));
		cvSetCaptureProperty(camRight, CV_CAP_PROP_FRAME_WIDTH, CAM_IMAGE_WIDTH);
		cvSetCaptureProperty(camRight, CV_CAP_PROP_FRAME_HEIGHT, CAM_IMAGE_HEIGHT);

		IplImage *imageLeft;
		IplImage *imageRight;
		float         fps = 0.f;
		int framecount = 0;
		long start = Platform::elapsedMillis();

		while (!terminateApp) {
			// wait on draw thread
				{
					unique_lock<mutex> lck(camMutex);
					while (!drawFinish && !terminateApp) { drawFinishCondition.wait(lck); }
				}

			imageLeft = cvQueryFrame(camLeft);
			if (imageLeft) {
				memcpy(perEyeWriteImage[0]->imageData, imageLeft->imageData, CAM_IMAGE_HEIGHT * CAM_IMAGE_WIDTH * 3);
			}
			else{
				SAY("Didn't get image of left cam");
			}

			imageRight = cvQueryFrame(camRight);
			if (imageRight) {
				memcpy(perEyeWriteImage[1]->imageData, imageRight->imageData, CAM_IMAGE_HEIGHT * CAM_IMAGE_WIDTH * 3);
			}
			else {
				SAY("Didn't get image of right cam");
			}

			long now = Platform::elapsedMillis();
			++framecount;
			if ((now - start) >= 2000) {
				float elapsed = (now - start) / 1000.f;
				fps = (float)framecount / elapsed;
				SAY("FPS cams: %0.2f\n", fps);
				start = now;
				framecount = 0;
			}

			{
				unique_lock<mutex> lck(camMutex);
				drawFinish = false;
				camFinish = true;
				camFinishCondition.notify_all();
			}

		}

		cvReleaseCapture(&camLeft);
		cvReleaseCapture(&camRight);
	}

void wait_for_cloud(uint n = 1) {
    // Wait until the cloud cb notifies us
    uint64_t old_cloud_count = cloud_count;
    unique_lock<std::mutex> unique_lock(cloud_count_lock);
    cloud_count_cv.wait(unique_lock, [=]{ return old_cloud_count + n <= cloud_count; });
}

// Thread function: Condition notifier
void ThreadCondition1(void * aArg)
{
  lock_guard<mutex> lock(gMutex);
  -- gCount;
  gCond.notify_all();
}

	void WakeUpThread() {
		_noWork.notify_one();
	}

	~QuadTreeThread() {
		_running = false;
		_noWork.notify_one();
	}