common/rfb/DecodeManager.cxx - android_external_tigervnc - Gitiles

 /* Copyright 2015 Pierre Ossman for Cendio AB
  *
  * This is free software; you can redistribute it and/or modify
  * it under the terms of the GNU General Public License as published by
  * the Free Software Foundation; either version 2 of the License, or
  * (at your option) any later version.
  *
  * This software is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
  *
  * You should have received a copy of the GNU General Public License
  * along with this software; if not, write to the Free Software
  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307,
  * USA.
  */

 #include <assert.h>
 #include <string.h>

 #include <rfb/CConnection.h>
 #include <rfb/DecodeManager.h>
 #include <rfb/Decoder.h>
 #include <rfb/Region.h>

 #include <rfb/LogWriter.h>

 #include <rdr/Exception.h>
 #include <rdr/MemOutStream.h>

 #include <os/Mutex.h>

 using namespace rfb;

 static LogWriter vlog("DecodeManager");

 DecodeManager::DecodeManager(CConnection *conn) :
   conn(conn), threadException(NULL)
 {
   size_t cpuCount;

   memset(decoders, 0, sizeof(decoders));

   queueMutex = new os::Mutex();
   producerCond = new os::Condition(queueMutex);
   consumerCond = new os::Condition(queueMutex);

   cpuCount = os::Thread::getSystemCPUCount();
   if (cpuCount == 0) {
     vlog.error("Unable to determine the number of CPU cores on this system");
     cpuCount = 1;
   } else {
     vlog.info("Detected %d CPU core(s)", (int)cpuCount);
     // No point creating more threads than this, they'll just end up
     // wasting CPU fighting for locks
     if (cpuCount > 4)
       cpuCount = 4;
     // The overhead of threading is small, but not small enough to
     // ignore on single CPU systems
     if (cpuCount == 1)
       vlog.info("Decoding data on main thread");
     else
       vlog.info("Creating %d decoder thread(s)", (int)cpuCount);
   }

   if (cpuCount == 1) {
     // Threads are not used on single CPU machines
     freeBuffers.push_back(new rdr::MemOutStream());
     return;
   }

   while (cpuCount--) {
     // Twice as many possible entries in the queue as there
     // are worker threads to make sure they don't stall
     freeBuffers.push_back(new rdr::MemOutStream());
     freeBuffers.push_back(new rdr::MemOutStream());

     threads.push_back(new DecodeThread(this));
   }
 }

 DecodeManager::~DecodeManager()
 {
   while (!threads.empty()) {
     delete threads.back();
     threads.pop_back();
   }

   delete threadException;

   while (!freeBuffers.empty()) {
     delete freeBuffers.back();
     freeBuffers.pop_back();
   }

   delete consumerCond;
   delete producerCond;
   delete queueMutex;

   for (size_t i = 0; i < sizeof(decoders)/sizeof(decoders[0]); i++)
     delete decoders[i];
 }

 void DecodeManager::decodeRect(const Rect& r, int encoding,
                                ModifiablePixelBuffer* pb)
 {
   Decoder *decoder;
   rdr::MemOutStream *bufferStream;

   QueueEntry *entry;

   assert(pb != NULL);

   if (!Decoder::supported(encoding)) {
     vlog.error("Unknown encoding %d", encoding);
     throw rdr::Exception("Unknown encoding");
   }

   if (!decoders[encoding]) {
     decoders[encoding] = Decoder::createDecoder(encoding);
     if (!decoders[encoding]) {
       vlog.error("Unknown encoding %d", encoding);
       throw rdr::Exception("Unknown encoding");
     }
   }

   decoder = decoders[encoding];

   // Fast path for single CPU machines to avoid the context
   // switching overhead
   if (threads.empty()) {
     bufferStream = freeBuffers.front();
     bufferStream->clear();
     decoder->readRect(r, conn->getInStream(), conn->server, bufferStream);
     decoder->decodeRect(r, bufferStream->data(), bufferStream->length(),
                         conn->server, pb);
     return;
   }

   // Wait for an available memory buffer
   queueMutex->lock();

   while (freeBuffers.empty())
     producerCond->wait();

   // Don't pop the buffer in case we throw an exception
   // whilst reading
   bufferStream = freeBuffers.front();

   queueMutex->unlock();

   // First check if any thread has encountered a problem
   throwThreadException();

   // Read the rect
   bufferStream->clear();
   decoder->readRect(r, conn->getInStream(), conn->server, bufferStream);

   // Then try to put it on the queue
   entry = new QueueEntry;

   entry->active = false;
   entry->rect = r;
   entry->encoding = encoding;
   entry->decoder = decoder;
   entry->server = &conn->server;
   entry->pb = pb;
   entry->bufferStream = bufferStream;

   decoder->getAffectedRegion(r, bufferStream->data(),
                              bufferStream->length(), conn->server,
                              &entry->affectedRegion);

   queueMutex->lock();

   // The workers add buffers to the end so it's safe to assume
   // the front is still the same buffer
   freeBuffers.pop_front();

   workQueue.push_back(entry);

   // We only put a single entry on the queue so waking a single
   // thread is sufficient
   consumerCond->signal();

   queueMutex->unlock();
 }

 void DecodeManager::flush()
 {
   queueMutex->lock();

   while (!workQueue.empty())
     producerCond->wait();

   queueMutex->unlock();

   throwThreadException();
 }

 void DecodeManager::setThreadException(const rdr::Exception& e)
 {
   os::AutoMutex a(queueMutex);

   if (threadException != NULL)
     return;

   threadException = new rdr::Exception("Exception on worker thread: %s", e.str());
 }

 void DecodeManager::throwThreadException()
 {
   os::AutoMutex a(queueMutex);

   if (threadException == NULL)
     return;

   rdr::Exception e(*threadException);

   delete threadException;
   threadException = NULL;

   throw e;
 }

 DecodeManager::DecodeThread::DecodeThread(DecodeManager* manager)
 {
   this->manager = manager;

   stopRequested = false;

   start();
 }

 DecodeManager::DecodeThread::~DecodeThread()
 {
   stop();
   wait();
 }

 void DecodeManager::DecodeThread::stop()
 {
   os::AutoMutex a(manager->queueMutex);

   if (!isRunning())
     return;

   stopRequested = true;

   // We can't wake just this thread, so wake everyone
   manager->consumerCond->broadcast();
 }

 void DecodeManager::DecodeThread::worker()
 {
   manager->queueMutex->lock();

   while (!stopRequested) {
     DecodeManager::QueueEntry *entry;

     // Look for an available entry in the work queue
     entry = findEntry();
     if (entry == NULL) {
       // Wait and try again
       manager->consumerCond->wait();
       continue;
     }

     // This is ours now
     entry->active = true;

     manager->queueMutex->unlock();

     // Do the actual decoding
     try {
       entry->decoder->decodeRect(entry->rect, entry->bufferStream->data(),
                                  entry->bufferStream->length(),
                                  *entry->server, entry->pb);
     } catch (rdr::Exception& e) {
       manager->setThreadException(e);
     } catch(...) {
       assert(false);
     }

     manager->queueMutex->lock();

     // Remove the entry from the queue and give back the memory buffer
     manager->freeBuffers.push_back(entry->bufferStream);
     manager->workQueue.remove(entry);
     delete entry;

     // Wake the main thread in case it is waiting for a memory buffer
     manager->producerCond->signal();
     // This rect might have been blocking multiple other rects, so
     // wake up every worker thread
     if (manager->workQueue.size() > 1)
       manager->consumerCond->broadcast();
   }

   manager->queueMutex->unlock();
 }

 DecodeManager::QueueEntry* DecodeManager::DecodeThread::findEntry()
 {
   std::list<DecodeManager::QueueEntry*>::iterator iter;
   Region lockedRegion;

   if (manager->workQueue.empty())
     return NULL;

   if (!manager->workQueue.front()->active)
     return manager->workQueue.front();

   for (iter = manager->workQueue.begin();
        iter != manager->workQueue.end();
        ++iter) {
     DecodeManager::QueueEntry* entry;

     std::list<DecodeManager::QueueEntry*>::iterator iter2;

     entry = *iter;

     // Another thread working on this?
     if (entry->active)
       goto next;

     // If this is an ordered decoder then make sure this is the first
     // rectangle in the queue for that decoder
     if (entry->decoder->flags & DecoderOrdered) {
       for (iter2 = manager->workQueue.begin(); iter2 != iter; ++iter2) {
         if (entry->encoding == (*iter2)->encoding)
           goto next;
       }
     }

     // For a partially ordered decoder we must ask the decoder for each
     // pair of rectangles.
     if (entry->decoder->flags & DecoderPartiallyOrdered) {
       for (iter2 = manager->workQueue.begin(); iter2 != iter; ++iter2) {
         if (entry->encoding != (*iter2)->encoding)
           continue;
         if (entry->decoder->doRectsConflict(entry->rect,
                                             entry->bufferStream->data(),
                                             entry->bufferStream->length(),
                                             (*iter2)->rect,
                                             (*iter2)->bufferStream->data(),
                                             (*iter2)->bufferStream->length(),
                                             *entry->server))
           goto next;
       }
     }

     // Check overlap with earlier rectangles
     if (!lockedRegion.intersect(entry->affectedRegion).is_empty())
       goto next;

     return entry;

 next:
     lockedRegion.assign_union(entry->affectedRegion);
   }

   return NULL;
 }
	/* Copyright 2015 Pierre Ossman for Cendio AB
	*
	* This is free software; you can redistribute it and/or modify
	* it under the terms of the GNU General Public License as published by
	* the Free Software Foundation; either version 2 of the License, or
	* (at your option) any later version.
	*
	* This software is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU General Public License for more details.
	*
	* You should have received a copy of the GNU General Public License
	* along with this software; if not, write to the Free Software
	* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307,
	* USA.
	*/

	#include <assert.h>
	#include <string.h>

	#include <rfb/CConnection.h>
	#include <rfb/DecodeManager.h>
	#include <rfb/Decoder.h>
	#include <rfb/Region.h>

	#include <rfb/LogWriter.h>

	#include <rdr/Exception.h>
	#include <rdr/MemOutStream.h>

	#include <os/Mutex.h>

	using namespace rfb;

	static LogWriter vlog("DecodeManager");

	DecodeManager::DecodeManager(CConnection *conn) :
	conn(conn), threadException(NULL)
	{
	size_t cpuCount;

	memset(decoders, 0, sizeof(decoders));

	queueMutex = new os::Mutex();
	producerCond = new os::Condition(queueMutex);
	consumerCond = new os::Condition(queueMutex);

	cpuCount = os::Thread::getSystemCPUCount();
	if (cpuCount == 0) {
	vlog.error("Unable to determine the number of CPU cores on this system");
	cpuCount = 1;
	} else {
	vlog.info("Detected %d CPU core(s)", (int)cpuCount);
	// No point creating more threads than this, they'll just end up
	// wasting CPU fighting for locks
	if (cpuCount > 4)
	cpuCount = 4;
	// The overhead of threading is small, but not small enough to
	// ignore on single CPU systems
	if (cpuCount == 1)
	vlog.info("Decoding data on main thread");
	else
	vlog.info("Creating %d decoder thread(s)", (int)cpuCount);
	}

	if (cpuCount == 1) {
	// Threads are not used on single CPU machines
	freeBuffers.push_back(new rdr::MemOutStream());
	return;
	}

	while (cpuCount--) {
	// Twice as many possible entries in the queue as there
	// are worker threads to make sure they don't stall
	freeBuffers.push_back(new rdr::MemOutStream());
	freeBuffers.push_back(new rdr::MemOutStream());

	threads.push_back(new DecodeThread(this));
	}
	}

	DecodeManager::~DecodeManager()
	{
	while (!threads.empty()) {
	delete threads.back();
	threads.pop_back();
	}

	delete threadException;

	while (!freeBuffers.empty()) {
	delete freeBuffers.back();
	freeBuffers.pop_back();
	}

	delete consumerCond;
	delete producerCond;
	delete queueMutex;

	for (size_t i = 0; i < sizeof(decoders)/sizeof(decoders[0]); i++)
	delete decoders[i];
	}

	void DecodeManager::decodeRect(const Rect& r, int encoding,
	ModifiablePixelBuffer* pb)
	{
	Decoder *decoder;
	rdr::MemOutStream *bufferStream;

	QueueEntry *entry;

	assert(pb != NULL);

	if (!Decoder::supported(encoding)) {
	vlog.error("Unknown encoding %d", encoding);
	throw rdr::Exception("Unknown encoding");
	}

	if (!decoders[encoding]) {
	decoders[encoding] = Decoder::createDecoder(encoding);
	if (!decoders[encoding]) {
	vlog.error("Unknown encoding %d", encoding);
	throw rdr::Exception("Unknown encoding");
	}
	}

	decoder = decoders[encoding];

	// Fast path for single CPU machines to avoid the context
	// switching overhead
	if (threads.empty()) {
	bufferStream = freeBuffers.front();
	bufferStream->clear();
	decoder->readRect(r, conn->getInStream(), conn->server, bufferStream);
	decoder->decodeRect(r, bufferStream->data(), bufferStream->length(),
	conn->server, pb);
	return;
	}

	// Wait for an available memory buffer
	queueMutex->lock();

	while (freeBuffers.empty())
	producerCond->wait();

	// Don't pop the buffer in case we throw an exception
	// whilst reading
	bufferStream = freeBuffers.front();

	queueMutex->unlock();

	// First check if any thread has encountered a problem
	throwThreadException();

	// Read the rect
	bufferStream->clear();
	decoder->readRect(r, conn->getInStream(), conn->server, bufferStream);

	// Then try to put it on the queue
	entry = new QueueEntry;

	entry->active = false;
	entry->rect = r;
	entry->encoding = encoding;
	entry->decoder = decoder;
	entry->server = &conn->server;
	entry->pb = pb;
	entry->bufferStream = bufferStream;

	decoder->getAffectedRegion(r, bufferStream->data(),
	bufferStream->length(), conn->server,
	&entry->affectedRegion);

	queueMutex->lock();

	// The workers add buffers to the end so it's safe to assume
	// the front is still the same buffer
	freeBuffers.pop_front();

	workQueue.push_back(entry);

	// We only put a single entry on the queue so waking a single
	// thread is sufficient
	consumerCond->signal();

	queueMutex->unlock();
	}

	void DecodeManager::flush()
	{
	queueMutex->lock();

	while (!workQueue.empty())
	producerCond->wait();

	queueMutex->unlock();

	throwThreadException();
	}

	void DecodeManager::setThreadException(const rdr::Exception& e)
	{
	os::AutoMutex a(queueMutex);

	if (threadException != NULL)
	return;

	threadException = new rdr::Exception("Exception on worker thread: %s", e.str());
	}

	void DecodeManager::throwThreadException()
	{
	os::AutoMutex a(queueMutex);

	if (threadException == NULL)
	return;

	rdr::Exception e(*threadException);

	delete threadException;
	threadException = NULL;

	throw e;
	}

	DecodeManager::DecodeThread::DecodeThread(DecodeManager* manager)
	{
	this->manager = manager;

	stopRequested = false;

	start();
	}

	DecodeManager::DecodeThread::~DecodeThread()
	{
	stop();
	wait();
	}

	void DecodeManager::DecodeThread::stop()
	{
	os::AutoMutex a(manager->queueMutex);

	if (!isRunning())
	return;

	stopRequested = true;

	// We can't wake just this thread, so wake everyone
	manager->consumerCond->broadcast();
	}

	void DecodeManager::DecodeThread::worker()
	{
	manager->queueMutex->lock();

	while (!stopRequested) {
	DecodeManager::QueueEntry *entry;

	// Look for an available entry in the work queue
	entry = findEntry();
	if (entry == NULL) {
	// Wait and try again
	manager->consumerCond->wait();
	continue;
	}

	// This is ours now
	entry->active = true;

	manager->queueMutex->unlock();

	// Do the actual decoding
	try {
	entry->decoder->decodeRect(entry->rect, entry->bufferStream->data(),
	entry->bufferStream->length(),
	*entry->server, entry->pb);
	} catch (rdr::Exception& e) {
	manager->setThreadException(e);
	} catch(...) {
	assert(false);
	}

	manager->queueMutex->lock();

	// Remove the entry from the queue and give back the memory buffer
	manager->freeBuffers.push_back(entry->bufferStream);
	manager->workQueue.remove(entry);
	delete entry;

	// Wake the main thread in case it is waiting for a memory buffer
	manager->producerCond->signal();
	// This rect might have been blocking multiple other rects, so
	// wake up every worker thread
	if (manager->workQueue.size() > 1)
	manager->consumerCond->broadcast();
	}

	manager->queueMutex->unlock();
	}

	DecodeManager::QueueEntry* DecodeManager::DecodeThread::findEntry()
	{
	std::list<DecodeManager::QueueEntry*>::iterator iter;
	Region lockedRegion;

	if (manager->workQueue.empty())
	return NULL;

	if (!manager->workQueue.front()->active)
	return manager->workQueue.front();

	for (iter = manager->workQueue.begin();
	iter != manager->workQueue.end();
	++iter) {
	DecodeManager::QueueEntry* entry;

	std::list<DecodeManager::QueueEntry*>::iterator iter2;

	entry = *iter;

	// Another thread working on this?
	if (entry->active)
	goto next;

	// If this is an ordered decoder then make sure this is the first
	// rectangle in the queue for that decoder
	if (entry->decoder->flags & DecoderOrdered) {
	for (iter2 = manager->workQueue.begin(); iter2 != iter; ++iter2) {
	if (entry->encoding == (*iter2)->encoding)
	goto next;
	}
	}

	// For a partially ordered decoder we must ask the decoder for each
	// pair of rectangles.
	if (entry->decoder->flags & DecoderPartiallyOrdered) {
	for (iter2 = manager->workQueue.begin(); iter2 != iter; ++iter2) {
	if (entry->encoding != (*iter2)->encoding)
	continue;
	if (entry->decoder->doRectsConflict(entry->rect,
	entry->bufferStream->data(),
	entry->bufferStream->length(),
	(*iter2)->rect,
	(*iter2)->bufferStream->data(),
	(*iter2)->bufferStream->length(),
	*entry->server))
	goto next;
	}
	}

	// Check overlap with earlier rectangles
	if (!lockedRegion.intersect(entry->affectedRegion).is_empty())
	goto next;

	return entry;

	next:
	lockedRegion.assign_union(entry->affectedRegion);
	}

	return NULL;
	}