void GBASyncUnlockAudio(struct GBASync* sync) {
	if (!sync) {
		return;
	}

	MutexUnlock(&sync->audioBufferMutex);
}

void UvdState::unlock()
{
    if (m_isRealtimeMode)
    {
        MutexUnlock(&m_lock);
    }
}

////////////////////////////////////////////////////////////
/// Destroy all users
////////////////////////////////////////////////////////////
bool UsersDestroy()
{
	struct UserData * Iterator = UserList.First;

	// Remove all users
	while (Iterator != NULL)
	{
		struct UserData * TempUser = Iterator;
		
		Iterator = Iterator->Next;

		// Close the socket
		MutexLock(&TempUser->MutexData);
		
		SocketClose(&TempUser->Connection.Socket);

		MutexUnlock(&TempUser->MutexData);

		// Destroy the mutex
		MutexDestroy(&TempUser->MutexData);

		// Free it
		free(TempUser);

		TempUser = NULL;
	}

	return true;
}

void* HeapAllocate(MemAllocHeap* heap, size_t size)
{
  bool thread_safe = 0 != (heap->m_Flags & HeapFlags::kThreadSafe);

  if (thread_safe)
  {
    MutexLock(&heap->m_Lock);
  }

  void* ptr = nullptr;
#if ENABLED(USE_DLMALLOC)
  ptr = mspace_malloc(heap->m_MemSpace, size);
#else
  ptr = malloc(size);
#endif
  if (!ptr)
  {
    Croak("out of memory allocating %d bytes", (int) size);
  }

  if (thread_safe)
  {
    MutexUnlock(&heap->m_Lock);
  }

  return ptr;
}

void* HeapReallocate(MemAllocHeap *heap, void *ptr, size_t size)
{
  bool thread_safe = 0 != (heap->m_Flags & HeapFlags::kThreadSafe);

  if (thread_safe)
  {
    MutexLock(&heap->m_Lock);
  }

  void *new_ptr;
#if ENABLED(USE_DLMALLOC)
  new_ptr = mspace_realloc(heap->m_MemSpace, ptr, size);
#else
  new_ptr = realloc(ptr, size);
#endif
  if (!new_ptr && size > 0)
  {
    Croak("out of memory reallocating %d bytes at %p", (int) size, ptr);
  }

  if (thread_safe)
  {
    MutexUnlock(&heap->m_Lock);
  }

  return new_ptr;
}

MUTEX * avtExecutionManager::FindMutex( const MUTEX_ID id )
{
    if (tPool == NULL)
        return NULL;

    std::map<MUTEX_ID, MUTEX *>::iterator it;
    MUTEX *lock;

    MutexLock( &mutexMapLock );

    it = mutexMap.find( id );
    if( it == mutexMap.end() )
    {
        // Not found, create it.
        lock = new MUTEX;
        MutexInit( lock );

        mutexMap.insert( std::pair<MUTEX_ID, MUTEX *>(id, lock) );
    }
    else
    {
        lock = it->second;
    }

    MutexUnlock( &mutexMapLock );

    return( lock );
}

int MQTTDisconnect(MQTTClient *c)
{
    int rc = FAILURE;
    Timer timer;     // we might wait for incomplete incoming publishes to complete
    int len = 0;

#if defined(MQTT_TASK)
    MutexLock(&c->mutex);
#endif
    TimerInit(&timer);
    TimerCountdownMS(&timer, c->command_timeout_ms);

    len = MQTTSerialize_disconnect(c->buf, c->buf_size);

    if (len > 0) {
        rc = sendPacket(c, len, &timer);    // send the disconnect packet
    }

    MQTTCloseSession(c);

#if defined(MQTT_TASK)
    MutexUnlock(&c->mutex);
#endif
    return rc;
}

  void BuildQueueDestroy(BuildQueue* queue)
  {
    Log(kDebug, "destroying build queue");
    const BuildQueueConfig* config = &queue->m_Config;

    MutexLock(&queue->m_Lock);
    queue->m_QuitSignalled = true;
    MutexUnlock(&queue->m_Lock);

    CondBroadcast(&queue->m_WorkAvailable);

    for (int i = 0, thread_count = config->m_ThreadCount; i < thread_count; ++i)
    {
      if (i > 0)
      {
        Log(kDebug, "joining with build thread %d", i);
        ThreadJoin(queue->m_Threads[i]);
      }

      ThreadStateDestroy(&queue->m_ThreadState[i]);
    }

    // Deallocate storage.
    MemAllocHeap* heap = queue->m_Config.m_Heap;
    HeapFree(heap, queue->m_ExpensiveWaitList);
    HeapFree(heap, queue->m_Queue);

    CondDestroy(&queue->m_WorkAvailable);
    MutexDestroy(&queue->m_Lock);

    // Unblock all signals on the main thread.
    SignalHandlerSetCondition(nullptr);
    SignalBlockThread(false);
  }

static THREAD_ENTRY _audioThread(void* context) {
	struct mPSP2AudioContext* audio = (struct mPSP2AudioContext*) context;
	uint32_t zeroBuffer[PSP2_SAMPLES] = {0};
	void* buffer = zeroBuffer;
	int audioPort = sceAudioOutOpenPort(SCE_AUDIO_OUT_PORT_TYPE_MAIN, PSP2_SAMPLES, 48000, SCE_AUDIO_OUT_MODE_STEREO);
	while (audio->running) {
		MutexLock(&audio->mutex);
		if (buffer != zeroBuffer) {
			// Can only happen in successive iterations
			audio->samples -= PSP2_SAMPLES;
			ConditionWake(&audio->cond);
		}
		if (audio->samples >= PSP2_SAMPLES) {
			buffer = &audio->buffer[audio->readOffset];
			audio->readOffset += PSP2_SAMPLES;
			if (audio->readOffset >= PSP2_AUDIO_BUFFER_SIZE) {
				audio->readOffset = 0;
			}
			// Don't mark samples as read until the next loop iteration to prevent
			// writing to the buffer while being read (see above)
		} else {
			buffer = zeroBuffer;
		}
		MutexUnlock(&audio->mutex);

		sceAudioOutOutput(audioPort, buffer);
	}
	sceAudioOutReleasePort(audioPort);
	return 0;
}

void GBASyncWaitFrameEnd(struct GBASync* sync) {
	if (!sync) {
		return;
	}

	MutexUnlock(&sync->videoFrameMutex);
}

void avtExecutionManager::MutexUnlock( const MUTEX_ID stringID )
{
    if (tPool == NULL)
        return;

    MutexUnlock( FindMutex(stringID) );
}

/**
 *	Call this function when a writer task finished using data that is
 *	protected by the given rwlock.
 */
void
rwl_writer_stop( TRWLock *rwlock )
{
	rwlock->wcount--;
        rwlock->wflag = 0;
        MutexUnlock( rwlock->datalock );
}

void KLogToScreenDisable(void)
{
	MutexLock (&dbg_mutex);

	klog_to_screen = FALSE;

	MutexUnlock (&dbg_mutex);
}

void GBASyncConsumeAudio(struct GBASync* sync) {
	if (!sync) {
		return;
	}

	ConditionWake(&sync->audioRequiredCond);
	MutexUnlock(&sync->audioBufferMutex);
}

bool GBASIOLockstepNodeUnload(struct GBASIODriver* driver) {
    struct GBASIOLockstepNode* node = (struct GBASIOLockstepNode*) driver;
    MutexLock(&node->p->mutex);
    --node->p->loaded;
    ConditionWake(&node->p->barrier);
    MutexUnlock(&node->p->mutex);
    return true;
}

void *PoolAlloc(TPool *p,uint32 size)
{
	TPoolZone *pz = 0;
	TPoolZone *npz = 0;
	void *x = 0;
	uint32 zonesize = 0;

	/* sanity */
	if (!p || !size) {
		return NULL;
	}

	if (!MutexLock(&p->mutex))
		return NULL;

	pz=p->currentzone;

	if (pz->pos+size<pz->maxpos)
	{
		x=pz->pos;
		pz->pos+=size;
		MutexUnlock(&p->mutex);
		return x;
	};

	if (size>p->zonesize)
		zonesize=size;
	else
		zonesize=p->zonesize;

	if( (npz = PoolZoneAlloc(zonesize)) )
	{
		npz->prev=pz;
		npz->next=pz->next;
		pz->next=npz;
		p->currentzone=npz; 
		x=(void *)(npz+1);
		npz->pos=((char *)(npz+1))+size;
	}
	else
		x=NULL;

	MutexUnlock(&p->mutex);
	return x;
}

static void _changeVideoSync(struct GBASync* sync, bool frameOn) {
	// Make sure the video thread can process events while the GBA thread is paused
	MutexLock(&sync->videoFrameMutex);
	if (frameOn != sync->videoFrameOn) {
		sync->videoFrameOn = frameOn;
		ConditionWake(&sync->videoFrameAvailableCond);
	}
	MutexUnlock(&sync->videoFrameMutex);
}

void GBASyncForceFrame(struct GBASync* sync) {
	if (!sync) {
		return;
	}

	MutexLock(&sync->videoFrameMutex);
	ConditionWake(&sync->videoFrameAvailableCond);
	MutexUnlock(&sync->videoFrameMutex);
}

int main()
{
	TThread loggingThread;
	TThread *loggerThreads = (TThread*)malloc(sizeof(TThread) * numThreadsToStart);
	unsigned int i = 0;

	printf("Creating threads\n");

	/* fire off the consumer thread */
	ThreadCreate(&loggingThread, log_error_thread, 0);

	sleep(1);

	/* fire off the producer threads */
	for (i=0; i<numThreadsToStart; ++i) {
		int *thrNum = (int*)malloc(sizeof(int));
		*thrNum = i;
		ThreadCreate(&loggerThreads[i], testLoggingThread, thrNum);
	}

	sleep(1);

	printf("Signalling START\n");

	/* signal that the threads can start */
	MutexLock(&cond_mutex);
	start_flag = 1;
	CondBroadcast(&start_cond);
	MutexUnlock(&cond_mutex);

	printf("Waiting for threads to complete\n");

	MutexLock(&threadCountLock);
	while (threadCount) {
		CondWait(&threadCountCond, &threadCountLock);
	}
	MutexUnlock(&threadCountLock);

	sleep(10);

	printf("See ya later, alligator\n");
	
	return EXIT_SUCCESS;
}

	ThreadFunc networkThread(void *) {
		while (true) {
			MutexLock(mutex);
			if (!threadRun) {
				MutexUnlock(mutex);
				break;
			}
			if (reqs.empty()) {
				MutexUnlock(mutex);
				continue;
			}
			Request& r = reqs.front();
			//if (r._signal == PLAYER_PACKET_SEND && ((PlayerPacket*)r._dataSend)->onlineID != player::onlineID)
			//	cout << "[ERROR]WTF!!!" << endl;
			if (r._dataSend != nullptr && r._dataLen != 0) {
				Net::Buffer buffer(r._dataLen + sizeof(int) * 2);
				int len = r._dataLen + sizeof(int);
				buffer.write((void*)&len, sizeof(int));
				buffer.write((void*)&r._signal, sizeof(int));
				buffer.write((void*)r._dataSend, r._dataLen);
				getClientSocket().send(buffer);
			}
			else {
				Net::Buffer buffer(sizeof(int) * 2);
				int len = sizeof(int);
				buffer.write((void*)&len, sizeof(int));
				buffer.write((void*)&r._signal, sizeof(int));
				getClientSocket().send(buffer);
			}
			if (r._callback) { //判断有无回调函数
				auto callback = r._callback;
				MutexUnlock(mutex);
				int len = getClientSocket().recvInt();   //获得数据长度
				Net::Buffer buffer(len);
				getClientSocket().recv(buffer, Net::BufferConditionExactLength(len));
				if (len > 0) callback(buffer.getData(), len); //调用回调函数
				MutexLock(mutex);
			}
			reqs.pop();
			MutexUnlock(mutex);
		}
		return 0;
	}