kresolvermanager.cpp

来自「konqueror3 embedded版本, KDE环境下的当家浏览器的嵌入式版」· C++ 代码 · 共 823 行 · 第 1/2 页

{
  /////
  // This function is called in a worker thread!!
  /////

  //qDebug("KResolverManager::releaseData(%u/%p): %p has been released", pid, 
//	 (void*)QThread::currentThread(), (void*)data);

  if (data->obj)
    {
      data->obj->status = KResolver::PostProcessing;	
    }
      
  data->worker->m_finished = true;
  data->worker->th = 0L;	// this releases the object

  // handle finished requests
  handleFinished();
}

// this function is called by KResolverManager::releaseData above
void KResolverManager::handleFinished()
{  
  bool redo = false;
  QPtrQueue<RequestData> doneRequests;

  mutex.lock();

  // loop over all items on the currently running list
  // we loop from the last to the first so that we catch requests with "requestors" before
  // we catch the requestor itself.
  RequestData *curr = currentRequests.last();
  while (curr)
    {
      if (curr->worker->th == 0L)
	{
	  if (handleFinishedItem(curr))
	    {
	      doneRequests.enqueue(currentRequests.take());
	      if (curr->requestor &&
		  curr->requestor->nRequests == 0 && 
		  curr->requestor->worker->m_finished)
		// there's a requestor that is now finished
		redo = true;
	    }
	}
      
      curr = currentRequests.prev();
    }
      
  //qDebug("KResolverManager::handleFinished(%u): %d requests to notify", pid, doneRequests.count());
  while (RequestData *d = doneRequests.dequeue())
    doNotifying(d);

  mutex.unlock();

  if (redo)
    {
      //qDebug("KResolverManager::handleFinished(%u): restarting processing to catch requestor",
	//     pid);
      handleFinished();
    }
}

// This function is called by KResolverManager::handleFinished above
bool KResolverManager::handleFinishedItem(RequestData* curr)
					  
{
  // for all items that aren't currently running, remove from the list
  // this includes all finished or cancelled requests

  if (curr->worker->m_finished && curr->nRequests == 0)
    {
      // this one has finished
      if (curr->obj)
	curr->obj->status = KResolver::PostProcessing; // post-processing is run in doNotifying()

      if (curr->requestor)
	--curr->requestor->nRequests;

      //qDebug("KResolverManager::handleFinishedItem(%u): removing %p since it's done",
	//     pid, (void*)curr);
      return true;
    }
  return false;
}



void KResolverManager::registerNewWorker(KResolverWorkerFactoryBase *factory)
{
  workerFactories.append(factory);
}

KResolverWorkerBase* KResolverManager::findWorker(KResolverPrivate* p)
{
  /////
  // this function can be called on any user thread
  /////

  // this function is called with an unlocked mutex and it's expected to be 
  // thread-safe!
  // but the factory list is expected not to be changed asynchronously

  // This function is responsible for finding a suitable worker for the given
  // input. That means we have to do a costly operation to create each worker
  // class and call their preprocessing functions. The first one that
  // says they can process (i.e., preprocess() returns true) will get the job.

  KResolverWorkerBase *worker;
  for (KResolverWorkerFactoryBase *factory = workerFactories.first(); factory; 
       factory = workerFactories.next())
    {
      worker = factory->create();

      // set up the data the worker needs to preprocess
      worker->input = &p->input;

      if (worker->preprocess())
	{
	  // good, this one says it can process
	  if (worker->m_finished)	   
	    p->status = KResolver::PostProcessing;
	  else
	    p->status = KResolver::Queued;
	  return worker;
	}

      // no, try again
      delete worker;
    }

  // found no worker
  return 0L;
}

void KResolverManager::doNotifying(RequestData *p)
{
  /////
  // This function may be called on any thread
  // any thread at all: user threads, GUI thread, manager thread or worker thread
  /////

  // Notification and finalisation
  //
  // Once a request has finished the normal processing, we call the
  // post processing function.
  //
  // After that is done, we will consolidate all results in the object's
  // KResolverResults and then post an event indicating that the signal
  // be emitted
  //
  // In case we detect that the object is waiting for completion, we do not
  // post the event, for KResolver::wait will take care of emitting the
  // signal.
  //
  // Once we release the mutex on the object, we may no longer reference it
  // for it might have been deleted.

  // "User" objects are those that are not created by the manager. Note that
  // objects created by worker threads are considered "user" objects. Objects
  // created by the manager are those created for KResolver::resolveAsync.
  // We should delete them.

  if (p->obj)
    {
      // lock the object
      p->obj->mutex.lock();
      KResolver* parent = p->obj->parent; // is 0 for non-"user" objects
      KResolverResults& r = p->obj->results;

      if (p->obj->status == KResolver::Canceled)
	{
	  p->obj->status = KResolver::Canceled;
	  p->obj->errorcode = KResolver::Canceled;
	  p->obj->syserror = 0;
	  r.setError(KResolver::Canceled, 0);
	}
      else if (p->worker)
	{
	  // post processing
	  p->worker->postprocess();	// ignore the result

	  // copy the results from the worker thread to the final
	  // object
	  r = p->worker->results;

	  // reset address
	  r.setAddress(p->input->node, p->input->service);

	  //qDebug("KResolverManager::doNotifying(%u/%p): for %p whose status is %d and has %d results", 
		 //pid, (void*)QThread::currentThread(), (void*)p, p->obj->status, r.count());

	  p->obj->errorcode = r.error();
	  p->obj->syserror = r.systemError();
	  p->obj->status = !r.isEmpty() ? 
	    KResolver::Success : KResolver::Failed;
	}
      else
	{
	  r.empty();
	  r.setError(p->obj->errorcode, p->obj->syserror);
	}

      // check whether there's someone waiting
      if (!p->obj->waiting && parent)
	// no, so we must post an event requesting that the signal be emitted
	// sorry for the C-style cast, but neither static nor reintepret cast work
	// here; I'd have to do two casts
	QApplication::postEvent(parent, new QEvent((QEvent::Type)(ResolutionCompleted)));

      // release the mutex
      p->obj->mutex.unlock();
    }
  else
    {
      // there's no object!
      if (p->worker)
	p->worker->postprocess();
    }

  delete p->worker;

  // ignore p->requestor and p->nRequests
  // they have been dealt with by the main loop

  delete p;

  // notify any objects waiting in KResolver::wait
  notifyWaiters.wakeAll();
}

// enqueue a new request
// this function is called from KResolver::start and 
// from KResolverWorkerBase::enqueue
void KResolverManager::enqueue(KResolver *obj, RequestData *requestor)
{
  RequestData *newrequest = new RequestData;
  newrequest->nRequests = 0;
  newrequest->obj = obj->d;
  newrequest->input = &obj->d->input;
  newrequest->requestor = requestor;

  // when processing a new request, find the most
  // suitable worker
  if ((newrequest->worker = findWorker(obj->d)) == 0L)
    {
      // oops, problem
      // cannot find a worker class for this guy
      obj->d->status = KResolver::Failed;
      obj->d->errorcode = KResolver::UnsupportedFamily;
      obj->d->syserror = 0;

      doNotifying(newrequest);
      return;
    }

  // no, queue it
  // p->status was set in findWorker!
  if (requestor)
    requestor->nRequests++;

  if (!newrequest->worker->m_finished)
    dispatch(newrequest);
  else if (newrequest->nRequests > 0)
    {
      mutex.lock();
      currentRequests.append(newrequest);
      mutex.unlock();
    }
  else
    // already done
    doNotifying(newrequest);
}

// a new request has been created
// dispatch it
void KResolverManager::dispatch(RequestData *data)
{
  // As stated in the beginning of the file, this function
  // is supposed to verify the availability of threads, start
  // any if necessary

  QMutexLocker locker(&mutex);

  // add to the queue
  newRequests.append(data);

  // check if we need to start a new thread
  //
  // we depend on the variables availableThreads and runningThreads to
  // know if we are supposed to start any threads:
  // - if availableThreads > 0, then there is at least one thread waiting,
  //    blocked in KResolverManager::requestData. It can't unblock
  //    while we are holding the mutex locked, therefore we are sure that
  //    our event will be handled
  // - if availableThreads == 0:
  //   - if runningThreads < maxThreads
  //     we will start a new thread, which will certainly block in
  //     KResolverManager::requestData because we are holding the mutex locked
  //   - if runningThreads == maxThreads
  //     This situation generally means that we have already maxThreads running
  //     and that all of them are processing. We will not start any new threads,
  //     but will instead wait for one to finish processing and request new data
  //
  //     There's a possible race condition here, which goes unhandled: if one of
  //     threads has timed out waiting for new data and is in the process of
  //     exiting. In that case, availableThreads == 0 and runningThreads will not
  //     have decremented yet. This means that we will not start a new thread
  //     that we could have. However, since there are other threads working, our
  //     event should be handled soon.
  //     It won't be handled if and only if ALL threads are in the process of 
  //     exiting. That situation is EXTREMELY unlikely and is not handled either.
  //
  if (availableThreads == 0 && runningThreads < maxThreads)
    {
      // yes, a new thread should be started

      // find if there's a finished one
      KResolverThread *th = workers.first();
      while (th && th->running())
	th = workers.next();

      if (th == 0L)
	// no, create one
	th = new KResolverThread;
      else
	workers.take();

      th->start();
      workers.append(th);
      runningThreads++;
    }

  feedWorkers.wakeAll();

  // clean up idle threads
  workers.first();
  while (workers.current())
    {
      if (!workers.current()->running())
	workers.remove();
      else
	workers.next();
    }
}

// this function is called by KResolverManager::dequeue
bool KResolverManager::dequeueNew(KResolver* obj)
{
  // This function must be called with a locked mutex
  // Deadlock warning:
  // always lock the global mutex first if both mutexes must be locked

  KResolverPrivate *d = obj->d;

  // check if it's in the new request list
  RequestData *curr = newRequests.first(); 
  while (curr)
    if (curr->obj == d)
      {
	// yes, this object is still in the list
	// but it has never been processed
	d->status = KResolver::Canceled;
	d->errorcode = KResolver::Canceled;
	d->syserror = 0;
	newRequests.take();

	delete curr->worker;
	delete curr;
	
	return true;
      }
    else
      curr = newRequests.next();

  // check if it's running
  curr = currentRequests.first();
  while (curr)
    if (curr->obj == d)
      {
	// it's running. We cannot simply take it out of the list.
	// it will be handled when the thread that is working on it finishes
	d->mutex.lock();

	d->status = KResolver::Canceled;
	d->errorcode = KResolver::Canceled;
	d->syserror = 0;

	// disengage from the running threads
	curr->obj = 0L;
	curr->input = 0L;
	if (curr->worker)
	  curr->worker->input = 0L;

	d->mutex.unlock();
      }
    else
      curr = currentRequests.next();

  return false;
}

// this function is called by KResolver::cancel
// it's expected to be thread-safe
void KResolverManager::dequeue(KResolver *obj)
{
  QMutexLocker locker(&mutex);
  dequeueNew(obj);
}