int main()
{
    WorkQueue q;
    work_t w = work_t(1, 2, 3, 4);
    q.add(w);
    assert(q.get().bot_y == w.bot_y);
    assert(q.get().id == 0);
    assert(q.remove(0).bot_y == w.bot_y);

    return 0;
}

DWORD WorkQueue::workThreadCallback(void* context)
{
    ASSERT_ARG(context, context);

    WorkQueue* queue = static_cast<WorkQueue*>(context);

    if (!queue->tryRegisterAsWorkThread())
        return 0;

    queue->performWorkOnRegisteredWorkThread();
    return 0;
}

void printWorkQueue(WorkQueue& queue) {
  WorkQueue local_queue(queue);
  printf("WorkQueue: %d elements\n", queue.size());
  for (long unsigned int request_index = 0;
          request_index < queue.size();
          ++request_index) {
    WorkRequest req = local_queue.top();
    local_queue.pop();
    printf("queue[%d]: node_index: %d, time_diff: %f\n", request_index,
            req.getIndex(), req.getTimeDiff());
  }
}

	//---------------------------------------------------------------------
	Page::Page(PageID pageID, PagedWorldSection* parent)
		: mID(pageID)
		, mParent(parent)
		, mDeferredProcessInProgress(false)
		, mModified(false)
		, mDebugNode(0)
	{
		WorkQueue* wq = Root::getSingleton().getWorkQueue();
		mWorkQueueChannel = wq->getChannel("Ogre/Page");
		wq->addRequestHandler(mWorkQueueChannel, this);
		wq->addResponseHandler(mWorkQueueChannel, this);
		touch();
	}

void Scheduler::renderFinished(WorkItem* item)
{
   WorkQueue* temp = priorityList.first();
   bool containsItem = false;
   if(item)
   {
      while((temp)&&(!containsItem))
      {
         containsItem = temp->renderFinished(item);
         temp = priorityList.next();
      }
   }
}

Octree::Octree(Context* context) :
    Component(context),
    Octant(BoundingBox(-DEFAULT_OCTREE_SIZE, DEFAULT_OCTREE_SIZE), 0, 0, this),
    numLevels_(DEFAULT_OCTREE_LEVELS)
{
    // Resize threaded ray query intermediate result vector according to number of worker threads
    WorkQueue* workQueue = GetSubsystem<WorkQueue>();
    rayQueryResults_.Resize(workQueue ? workQueue->GetNumThreads() + 1 : 1);

    // If the engine is running headless, subscribe to RenderUpdate events for manually updating the octree
    // to allow raycasts and animation update
    if (!GetSubsystem<Graphics>())
        SubscribeToEvent(E_RENDERUPDATE, HANDLER(Octree, HandleRenderUpdate));
}

	//------------------------------------------------------------------------
	BackgroundProcessTicket ResourceBackgroundQueue::addRequest(ResourceRequest& req)
	{
		WorkQueue* queue = Root::getSingleton().getWorkQueue();

		Any data(req);

		WorkQueue::RequestID requestID = 
			queue->addRequest(mWorkQueueChannel, (uint16)req.type, data);


		mOutstandingRequestSet.insert(requestID);

		return requestID;
	}

TEST(WorkQueue, cancel) {
    RunLoop loop(uv_default_loop());

    WorkQueue queue;

    auto work = [&]() {
        FAIL() << "Should never be called";
    };

    queue.push(work);
    queue.push(work);
    queue.push(work);
    queue.push(work);
    queue.push(work);
}

void MethodTransform::sync_all() {
  std::vector<MethodTransform*> transforms;
  for (auto& centry : s_cache) {
    transforms.push_back(centry.second);
  }
  std::vector<WorkItem<MethodTransform>> workitems(transforms.size());
  auto mt_sync = [](MethodTransform* mt) { mt->sync(); };
  for (size_t i = 0; i < transforms.size(); i++) {
    workitems[i].init(mt_sync, transforms[i]);
  }
  WorkQueue wq;

  if (workitems.size() > 0) {
    wq.run_work_items(&workitems[0], (int)workitems.size());
  }
}

    StatusWith<ReplicationExecutor::CallbackHandle> ReplicationExecutor::scheduleWorkAt(
            Date_t when,
            const CallbackFn& work) {

        boost::lock_guard<boost::mutex> lk(_mutex);
        WorkQueue temp;
        StatusWith<CallbackHandle> cbHandle = enqueueWork_inlock(&temp, work);
        if (!cbHandle.isOK())
            return cbHandle;
        cbHandle.getValue()._iter->readyDate = when;
        WorkQueue::iterator insertBefore = _sleepersQueue.begin();
        while (insertBefore != _sleepersQueue.end() && insertBefore->readyDate <= when)
            ++insertBefore;
        _sleepersQueue.splice(insertBefore, temp, temp.begin());
        return cbHandle;
    }

void test_work_queue() {
  WorkQueue<string> q;
  string t = "hello";
  q.push(t);
  string s;
  if(q.try_pop(s, 500)){
    cout << "got " << s << " from queue" << endl;
  }
  bool ok = q.try_pop(s,500);
  if(ok) {
    cout << "failed, should have timed out" << endl;
  }
  else{
    cout << "passed, timed out as expected" << endl;
  }
}

static void didCloseOnConnectionWorkQueue(WorkQueue& workQueue, CoreIPC::Connection*)
{
    // If the connection has been closed and we haven't responded in the main thread for 10 seconds
    // the process will exit forcibly.
    const double watchdogDelay = 10;

    workQueue.dispatchAfterDelay(bind(static_cast<void(*)()>(watchdogCallback)), watchdogDelay);
}

void ChildProcess::didCloseOnConnectionWorkQueue(WorkQueue& workQueue, CoreIPC::Connection*)
{
    // If the connection has been closed and we haven't responded in the main thread for 10 seconds
    // the process will exit forcibly.
    static const double watchdogDelay = 10.0;
    
    workQueue.scheduleWorkAfterDelay(WorkItem::create(watchdogCallback), watchdogDelay);
}

    //---------------------------------------------------------------------
    TerrainPagedWorldSection::TerrainPagedWorldSection(const String& name, PagedWorld* parent, SceneManager* sm)
        : PagedWorldSection(name, parent, sm)
        , mTerrainGroup(0)
        , mTerrainDefiner(0)
        , mHasRunningTasks(false)
        , mLoadingIntervalMs(900)
    {
        // we always use a grid strategy
        setStrategy(parent->getManager()->getStrategy("Grid2D"));

        WorkQueue* wq = Root::getSingleton().getWorkQueue();
        mWorkQueueChannel = wq->getChannel("Ogre/TerrainPagedWorldSection");
        wq->addRequestHandler(mWorkQueueChannel, this);
        wq->addResponseHandler(mWorkQueueChannel, this);

        mNextLoadingTime = Root::getSingletonPtr()->getTimer()->getMilliseconds();
    }

Results FastqWriterThread(WorkQueue<Results>& queue, const string& fname)
{
    ofstream ccsFastq(fname);
    Results counts;
    while (queue.ConsumeWith(WriteFastqRecords, ref(ccsFastq), ref(counts)))
        ;
    return counts;
}

Results BamWriterThread(WorkQueue<Results>& queue, unique_ptr<BamWriter>&& ccsBam,
                        unique_ptr<PbiBuilder>&& ccsPbi)
{
    Results counts;
    while (queue.ConsumeWith(WriteBamRecords, ref(*ccsBam), ref(ccsPbi), ref(counts)))
        ;
    return counts;
}

//--------------------------------------------------------------------------
void MeshLodTests::blockedWaitForLodGeneration(const MeshPtr& mesh)
{
    bool success = false;
    const int timeout = 5000;
    WorkQueue* wq = Root::getSingleton().getWorkQueue();
    for (int i = 0; i < timeout; i++) 
    {
        OGRE_THREAD_SLEEP(1);
        wq->processResponses(); // Injects the Lod if ready
        if (mesh->getNumLodLevels() != 1) {
            success = true;
            break;
        }
    }
    // timeout
    CPPUNIT_ASSERT(success);
}

// store equivalence class in VertexInfo for each vertex
static
vector<VertexInfoSet> partitionGraph(ptr_vector<VertexInfo> &infos,
                                     WorkQueue &work_queue, const NGHolder &g,
                                     EquivalenceType eq) {
    const size_t num_verts = infos.size();

    vector<VertexInfoSet> classes;
    unordered_map<ClassInfo, unsigned> classinfomap;

    // assume we will have lots of classes, so we don't waste time resizing
    // these structures.
    classes.reserve(num_verts);
    classinfomap.reserve(num_verts);

    // get distances from start (or accept) for all vertices
    // only one of them is used at a time, never both
    vector<NFAVertexDepth> depths;
    vector<NFAVertexRevDepth> rdepths;

    if (eq == LEFT_EQUIVALENCE) {
        calcDepths(g, depths);
    } else {
        calcDepths(g, rdepths);
    }

    // partition the graph based on CharReach
    for (VertexInfo &vi : infos) {
        ClassInfo::ClassDepth depth;

        if (eq == LEFT_EQUIVALENCE) {
            depth = depths[vi.vert_index];
        } else {
            depth = rdepths[vi.vert_index];
        }
        ClassInfo ci(g, vi, depth, eq);

        auto ii = classinfomap.find(ci);
        if (ii == classinfomap.end()) {
            // vertex is in a new equivalence class by itself.
            unsigned eq_class = classes.size();
            vi.equivalence_class = eq_class;
            classes.push_back({&vi});
            classinfomap.emplace(move(ci), eq_class);
        } else {
            // vertex is added to an existing class.
            unsigned eq_class = ii->second;
            vi.equivalence_class = eq_class;
            classes.at(eq_class).insert(&vi);

            // we now know that this particular class has more than one
            // vertex, so we add it to the work queue
            work_queue.push(eq_class);
        }
    }

    DEBUG_PRINTF("partitioned, %zu equivalence classes\n", classes.size());
    return classes;
}

void *worker(void *vtp)
{
    fprintf(stderr, "Worker working\n");
    WorkQueue<Task> *tqp = (WorkQueue<Task> *)vtp;
    Task tsk;
    for (;;) {
	if (!tqp->take(&tsk)) {
	    fprintf(stderr, "Worker: take failed\n");
	    return (void*)0;
	}
	fprintf(stderr, "WORKER: got task %d\n", tsk.m_id);
	if (tsk.m_id > 20) {
	    tqp->workerExit();
	    break;
	}
    }
    return (void*)1;
}

	//---------------------------------------------------------------------
	Page::~Page()
	{
		WorkQueue* wq = Root::getSingleton().getWorkQueue();
		wq->removeRequestHandler(mWorkQueueChannel, this);
		wq->removeResponseHandler(mWorkQueueChannel, this);

		destroyAllContentCollections();
		if (mDebugNode)
		{
			// destroy while we have the chance
			SceneNode::ObjectIterator it = mDebugNode->getAttachedObjectIterator();
			while(it.hasMoreElements())
				mParent->getSceneManager()->destroyMovableObject(it.getNext());
			mDebugNode->removeAndDestroyAllChildren();
			mParent->getSceneManager()->destroySceneNode(mDebugNode);

			mDebugNode = 0;
		}
	}