void AsyncDecodeWebAudio(const char* aContentType, uint8_t* aBuffer,
                         uint32_t aLength, WebAudioDecodeJob& aDecodeJob) {
  Maybe<MediaContainerType> containerType =
      MakeMediaContainerType(aContentType);
  // Do not attempt to decode the media if we were not successful at sniffing
  // the container type.
  if (!*aContentType || strcmp(aContentType, APPLICATION_OCTET_STREAM) == 0 ||
      !containerType) {
    nsCOMPtr<nsIRunnable> event =
        new ReportResultTask(aDecodeJob, &WebAudioDecodeJob::OnFailure,
                             WebAudioDecodeJob::UnknownContent);
    JS_free(nullptr, aBuffer);
    aDecodeJob.mContext->Dispatch(event.forget());
    return;
  }

  RefPtr<MediaDecodeTask> task =
      new MediaDecodeTask(*containerType, aBuffer, aLength, aDecodeJob);
  if (!task->CreateReader()) {
    nsCOMPtr<nsIRunnable> event =
        new ReportResultTask(aDecodeJob, &WebAudioDecodeJob::OnFailure,
                             WebAudioDecodeJob::UnknownError);
    aDecodeJob.mContext->Dispatch(event.forget());
  } else {
    // If we did this without a temporary:
    //   task->Reader()->OwnerThread()->Dispatch(task.forget())
    // we might evaluate the task.forget() before calling Reader(). Enforce
    // a non-crashy order-of-operations.
    TaskQueue* taskQueue = task->Reader()->OwnerThread();
    nsresult rv = taskQueue->Dispatch(task.forget());
    MOZ_DIAGNOSTIC_ASSERT(NS_SUCCEEDED(rv));
    Unused << rv;
  }
}

int main( int argc, char* argv[] )
{
	// Our progress counter.
	auto counter = pc::ProgressCounter::Create();
	// Our tasks queue.
	TaskQueue tasks;

	// Produce the tasks.
	std::mt19937 rnd_device( (uint32_t)time( 0 ) );
	for( int32_t task_id = 0; task_id < 680; ++task_id )
	{
		auto goal = counter->ProduceGoal();
		std::chrono::milliseconds interval{ 600 + rnd_device() % 1400 };
		tasks.PushTask( { [goal, interval](){ std::this_thread::sleep_for( interval ); } } );
	};

	// Produce the workers.
	tasks.Lock();
	size_t threads_count = std::thread::hardware_concurrency();
	while( threads_count-- > 0 )
	{
		std::thread t( ThreadFunction, std::ref( tasks ) );
		t.detach();
	};
	tasks.Unlock();

	// Display the progress of tasks completion until the progress reaches 100%
	do
	{
		printf( "\rCurrent progress : %3.3f%%; Tasks remained : %zd;", 100.0f * counter->GetFinisedPercent(), tasks.GetCount() );
		std::this_thread::sleep_for( std::chrono::milliseconds( 500 ) );
	}
	while( counter->GetRemainsPercent() > 0.0f );
};

int loadQueue(const char *pPath, TaskQueue &queue)
{
    DIR *pDir = opendir(pPath);
    if (!pDir){
        if (errno == ENOTDIR && access(pPath, F_OK) == 0) {
            char *pFile = new char[strlen(pPath) + 1];
            strcpy(pFile, pPath);
            queue.pushBack(pFile);
            return 0;
        }
        return -1;
    }

    struct dirent *de = NULL;
    while ((de=readdir(pDir))!=NULL) {
        if (de->d_type != DT_REG && de->d_type != DT_LNK) {
            continue;
        }
        char *pFile = new char[strlen(pPath) + strlen(de->d_name) + 2];
        snprintf(pFile, strlen(pPath) + strlen(de->d_name) + 2, "%s/%s", pPath, de->d_name);
        queue.pushBack(pFile);
        TNOTE("find file %s", pFile);
    }
    closedir(pDir);
    return 0;
}

void IDBTransactionBackendImpl::taskTimerFired(Timer<IDBTransactionBackendImpl>*)
{
    IDB_TRACE("IDBTransactionBackendImpl::taskTimerFired");
    ASSERT(!isTaskQueueEmpty());

    if (m_state == StartPending) {
        m_transaction.begin();
        m_state = Running;
    }

    // The last reference to this object may be released while performing the
    // tasks. Take take a self reference to keep this object alive so that
    // the loop termination conditions can be checked.
    RefPtr<IDBTransactionBackendImpl> protect(this);

    TaskQueue* taskQueue = m_pendingPreemptiveEvents ? &m_preemptiveTaskQueue : &m_taskQueue;
    while (!taskQueue->isEmpty() && m_state != Finished) {
        ASSERT(m_state == Running);
        OwnPtr<Operation> task(taskQueue->takeFirst());
        task->perform(this);

        // Event itself may change which queue should be processed next.
        taskQueue = m_pendingPreemptiveEvents ? &m_preemptiveTaskQueue : &m_taskQueue;
    }

    // If there are no pending tasks, we haven't already committed/aborted,
    // and the front-end requested a commit, it is now safe to do so.
    if (!hasPendingTasks() && m_state != Finished && m_commitPending)
        commit();
}

map<uint32_t, NodeList> Worker::get_map(TaskQueue &mqueue) {
	map<uint32_t, NodeList> update_map;
	/*Package package;
	 package.set_operation(operation);
	 if(operation == 25) {
	 package.set_currnode(toid);
	 }*/
	uint32_t num_nodes = svrclient.memberList.size();
	for (TaskQueue::iterator it = mqueue.begin(); it != mqueue.end(); ++it) {
		uint32_t serverid = myhash(((*it)->task_id).c_str(), num_nodes);
		string str((*it)->task_id);
		str.append("\'");
		if (update_map.find(serverid) == update_map.end()) {
			str.append("\"");
			NodeList new_list;
			new_list.push_back(str);
			update_map.insert(make_pair(serverid, new_list));
		} else {
			NodeList &exist_list = update_map[serverid];
			string last_str(exist_list.back());
			if ((last_str.size() + str.size()) > STRING_THRESHOLD) {
				str.append("\"");
				exist_list.push_back(str);
			} else {
				exist_list.pop_back();
				str.append(last_str);
				exist_list.push_back(str);
			}
		}
	}
	return update_map;
}

void
AsyncDecodeWebAudio(const char* aContentType, uint8_t* aBuffer,
                    uint32_t aLength, WebAudioDecodeJob& aDecodeJob)
{
  // Do not attempt to decode the media if we were not successful at sniffing
  // the content type.
  if (!*aContentType ||
      strcmp(aContentType, APPLICATION_OCTET_STREAM) == 0) {
    nsCOMPtr<nsIRunnable> event =
      new ReportResultTask(aDecodeJob,
                           &WebAudioDecodeJob::OnFailure,
                           WebAudioDecodeJob::UnknownContent);
    JS_free(nullptr, aBuffer);
    NS_DispatchToMainThread(event);
    return;
  }

  RefPtr<MediaDecodeTask> task =
    new MediaDecodeTask(aContentType, aBuffer, aLength, aDecodeJob);
  if (!task->CreateReader()) {
    nsCOMPtr<nsIRunnable> event =
      new ReportResultTask(aDecodeJob,
                           &WebAudioDecodeJob::OnFailure,
                           WebAudioDecodeJob::UnknownError);
    NS_DispatchToMainThread(event);
  } else {
    // If we did this without a temporary:
    //   task->Reader()->OwnerThread()->Dispatch(task.forget())
    // we might evaluate the task.forget() before calling Reader(). Enforce
    // a non-crashy order-of-operations.
    TaskQueue* taskQueue = task->Reader()->OwnerThread();
    taskQueue->Dispatch(task.forget());
  }
}

void TaskScheduler::updateFramesClass(TaskQueue& queue, Class* klass)
{
	for(TaskQueue::iterator it = queue.begin(); it != queue.end(); it++)
	{
		Task* task = *it;
		task->getContext()->updateFramesClass(klass);
	}
}

int main()
{
    TaskQueue q;
    q.post([]{ std::cout << "Hello World!" << std::endl; });
    q.post([]{ std::cout << "Hello World!" << std::endl; });
    q.poll_one();
    q.post([]{ std::cout << "Hello World!" << std::endl; });
    q.poll_all();
}

void ThreadFunction( TaskQueue& tasks )
{
	while( !tasks.IsEmpty() )
	{
		{
			auto task = tasks.PullTask();
			// Process this task.
			task();
		};
	};
};

void HTMLConstructionSite::executeQueuedTasks()
{
    const size_t size = m_taskQueue.size();
    if (!size)
        return;

    // Copy the task queue into a local variable in case executeTask
    // re-enters the parser.
    TaskQueue queue;
    queue.swap(m_taskQueue);

    for (size_t i = 0; i < size; ++i)
        executeTask(queue[i]);

    // We might be detached now.
}

int main() {
    TaskQueue<FibonacciTask> fibonacciTasks;
    for (int i = 42; i >= 39; --i) {
        fibonacciTasks.add([i](size_t threadId) { runFibonacci(i, threadId); });
    }

    ThreadPool<FibonacciTask> threadPool(4, fibonacciTasks);

    for (int j = 0; j < 5; ++j) {
        for (int i = 35; i > 20; --i) {
            fibonacciTasks.add([i](size_t threadId) { runFibonacci(i, threadId); });
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    }
    return 0;
}

        void run() {
            tot = 0;
            TaskQueue<V> d;
            int x = 0;
            for( int i = 0; i < 100; i++ ) {
                if( i % 30 == 0 )
                    d.invoke();

                x += i;
                writelock lk;
                V v;
                v.val = i;
                d.defer(v);
            }
            d.invoke();
            assert( x == tot );
        }

 INLINE void ActivateOneOutputTask(TaskQueue& queue, int task, int* tasknum)
 {   
     if (DEC_ATOMIC(&fTaskList[task]) == 0) {
         *tasknum = task;
     } else {
         *tasknum = queue.PopHead(); 
     }
 }

int Worker::move_task_to_ready_queue(TaskQueue_Item **qi) {
	//pthread_mutex_lock(&w_lock);
	pthread_mutex_lock(&lock);
	rqueue.push_back(*qi);
	//wqueue.erase(*qi);
	pthread_mutex_unlock(&lock);
	//pthread_mutex_unlock(&w_lock);
}

void fun1(void* data) {
  TaskQueue* queue = static_cast<TaskQueue*>(data);
  cout<< " thread "<< std::this_thread::get_id()<<" run" << endl;
  std::chrono::milliseconds dura( 5);
  cout.flush();
  for (int i = 0; i < 10; i++) {
    Task* task = new DemonTask();
    queue->PushBack(task);
    //std::this_thread::sleep_for(dura);
  }
  cout<<"f1:queue size" << queue->size()<<endl;;
  cout.flush();
  Task* task = queue->PopFront();
  task->Action();
  //task->Release();
  cout<<"f1:queue size" << queue->size()<<endl;;
  cout.flush();
}

 INLINE void ActivateOutputTask(TaskQueue& queue, int task, int* tasknum)
 {
     if (DEC_ATOMIC(&fTaskList[task]) == 0) {
         if (*tasknum == WORK_STEALING_INDEX) {
             *tasknum = task;
         } else {
             queue.PushHead(task);
         }
     }    
 }

void IDBTransactionBackendImpl::taskTimerFired(Timer<IDBTransactionBackendImpl>*)
{
    IDB_TRACE("IDBTransactionBackendImpl::taskTimerFired");
    ASSERT(!m_taskQueue.isEmpty());

    if (m_state == StartPending) {
        m_transaction->begin();
        m_state = Running;
    }

    TaskQueue queue;
    queue.swap(m_taskQueue);
    while (!queue.isEmpty() && m_state != Finished) {
        ASSERT(m_state == Running);
        OwnPtr<ScriptExecutionContext::Task> task(queue.first().release());
        queue.removeFirst();
        m_pendingEvents++;
        task->performTask(0);
    }
}

/*
* read_tasks_from_file : reads the table information from a text file.
* Parameters:
* - file_name : the file containing the table definitions
* - count_only : indicates if the file contains information to count the records
*                from the source DB or to actually trasmit the data
* - tasks : output parameter that will contain a task for each table definition loaded
*           from the file
  - resume : indicates if the file contains information to resume copying data from
             last PK
* - max_count : limit copied rows count to max_count
*
* Remarks : Each table is defined in a single line with the next format for count_only = true and resume = false
*           <src_schema>\t<src_table>\n
*
*           and in the next format for a count_only = false or resume = true
*           <src_schema>\t<src_table>\t<tgt_schema>\t<tgt_table>\t<source_pk_columns>\t<target_pk_columns>\t<select_expression>
*/
bool read_tasks_from_file(const std::string file_name, bool count_only, TaskQueue& tasks, std::set<std::string> &trigger_schemas,
                          bool resume, long long int max_count)
{
  std::ifstream ifs ( file_name.data() , std::ifstream::in );
  unsigned int field_count = count_only && !resume ? 2 : 7;
  bool error = false;

  printf("Loading table information from file %s\n", file_name.data());

  while (!error && ifs.good())
  {
    TableParam param;
    std::string line;
    getline(ifs, line);

    if (line.length())
    {
      log_info("--table %s\n", line.data());

      std::vector<std::string> fields = base::split(line, "\t", field_count);

      if (fields.size() == field_count)
      {
        param.source_schema = fields[0];
        param.source_table = fields[1];

        if (!(count_only && !resume))
        {
          param.target_schema = fields[2];
          param.target_table = fields[3];
          if(std::strcmp(fields[4].c_str(), "-") != 0)
            param.source_pk_columns = base::split(fields[4], ",", -1);
          if(std::strcmp(fields[5].c_str(), "-") != 0)
            param.target_pk_columns = base::split(fields[5], ",", -1);
          param.select_expression = fields[6];

          trigger_schemas.insert(param.target_schema);
        }

        param.copy_spec.resume = resume;
        param.copy_spec.max_count = max_count;
        param.copy_spec.type = CopyAll;
        tasks.add_task(param);
      }
      else
        error = true;
    }
  }

  ifs.close();

  return !error;
}

void* check_wait_queue(void* args) {
	Worker *worker = (Worker*) args;
	TaskQueue_Item *qi;
	while (ON) {
		while (wqueue.size() > 0) {
			//for(TaskQueue::iterator it = wqueue.begin(); it != wqueue.end(); ++it) {
			int size = wqueue.size();
			for (int i = 0; i < size; i++) {
				//qi = *it;
				qi = wqueue[i];
				if (qi != NULL) {
					int status = worker->check_if_task_is_ready(qi->task_id); //cout << "task = " << qi->task_id << " status = " << status << endl;
					if (status == 0) {
						//cout << "task = " << qi->task_id << " status = " << status << endl;
						int ret = worker->move_task_to_ready_queue(&qi);
						pthread_mutex_lock(&w_lock);
						wqueue[i] = NULL;
						pthread_mutex_unlock(&w_lock);
					}
					/*if(status < 0) {
					 cout << "negative numwait" << endl;
					 }*/
				}
			}
			pthread_mutex_lock(&w_lock);
			TaskQueue::iterator last = remove_if(wqueue.begin(), wqueue.end(),
					check);
			wqueue.erase(last, wqueue.end());
			pthread_mutex_unlock(&w_lock);
			sleep(1);
		}
	}
}

int front_task_to_string(char * buffer)
{
    if (NULL == buffer)
    {
        return -1;
    }

    if (qu.empty())
    {
        return -2;
    }
    
    TaskPtr p = qu.front();

    if (p == NULL)
    {
        return -3;
    }

    int s = (int)p->ts;

    sprintf(buffer, "<Tasks>"
        "<task>"
        "<id>%d</id>"
        "<name>%s</name>"
        "<sid>%d</sid>"
        "<subdir>%s</subdir>"
        "<status>%s</status>"
        "</task>"
        "</Tasks>",
        p->id, p->name, p->sid, p->subdir, TaskStateText[s]);

    return 0;
}