Index: kio/kio/job.cpp
===================================================================
--- kio/kio/job.cpp	(revision 1024592)
+++ kio/kio/job.cpp	(working copy)
@@ -308,9 +308,9 @@
     }
 
     Scheduler::doJob(q);
+    Scheduler::scheduleJob(q);
 }
 
-
 bool SimpleJob::doKill()
 {
     //kDebug() << this;
@@ -627,6 +627,7 @@
         // Return slave to the scheduler
         d->slaveDone();
         Scheduler::doJob(this);
+        Scheduler::scheduleJob(this);
     }
 }
 
@@ -834,6 +835,7 @@
         // Return slave to the scheduler
         d->slaveDone();
         Scheduler::doJob(this);
+        Scheduler::scheduleJob(this);
     }
 }
 
@@ -999,6 +1001,7 @@
         // Return slave to the scheduler
         d->slaveDone();
         Scheduler::doJob(this);
+        Scheduler::scheduleJob(this);
     }
 }
 
@@ -1597,6 +1600,7 @@
         // Return slave to the scheduler
         d->slaveDone();
         Scheduler::doJob(this);
+        Scheduler::scheduleJob(this);
     }
 }
 
@@ -1803,6 +1807,8 @@
    else
    {
       startDataPump();
+      if (m_putJob)
+        SimpleJobPrivate::get(m_putJob)->m_pairedUrl = m_src;
    }
 }
 
@@ -2420,6 +2426,7 @@
         // Return slave to the scheduler
         d->slaveDone();
         Scheduler::doJob(this);
+        Scheduler::scheduleJob(this);
     }
 }
 
@@ -2671,6 +2678,7 @@
         d->m_url = d->m_waitQueue.first().url;
         d->slaveDone();
         Scheduler::doJob(this);
+        Scheduler::scheduleJob(this);
      }
   }
 }
Index: kio/kio/job_p.h
===================================================================
--- kio/kio/job_p.h	(revision 1024592)
+++ kio/kio/job_p.h	(working copy)
@@ -98,6 +98,13 @@
         KUrl m_url;
         KUrl m_subUrl;
         int m_command;
+        /*
+          In high-level jobs such as FileCopyJob, this variable represents the
+          source (GET) url and used by KIO::Scheduler to avoid deadlock conditions
+          when scheduling jobs with two remote ends, e.g. copying file from one ftp
+          server to another one.
+         */
+        KUrl m_pairedUrl;
 
         void simpleJobInit();
 
Index: kio/kio/scheduler.cpp
===================================================================
--- kio/kio/scheduler.cpp	(revision 1024592)
+++ kio/kio/scheduler.cpp	(working copy)
@@ -55,6 +55,11 @@
     return SimpleJobPrivate::get(job)->m_command;
 }
 
+static inline KUrl pairedRequest(SimpleJob *job)
+{
+    return SimpleJobPrivate::get(job)->m_pairedUrl;
+}
+
 static inline void startJob(SimpleJob *job, Slave *slave)
 {
     SimpleJobPrivate::get(job)->start(slave);
@@ -136,8 +141,8 @@
     void publishSlaveOnHold();
     void registerWindow(QWidget *wid);
 
-    Slave *findIdleSlave(ProtocolInfo *protInfo, SimpleJob *job, bool &exact);
-    Slave *createSlave(ProtocolInfo *protInfo, SimpleJob *job, const KUrl &url);
+    Slave *findIdleSlave(ProtocolInfo *protInfo, SimpleJob *job, bool &exact, bool enforeLimits = false);
+    Slave *createSlave(ProtocolInfo *protInfo, SimpleJob *job, const KUrl& url, bool enforceLimits = false);
 
     void debug_info();
 
@@ -162,7 +167,7 @@
 class KIO::SchedulerPrivate::ProtocolInfo
 {
 public:
-    ProtocolInfo() : maxSlaves(1), skipCount(0)
+    ProtocolInfo() : maxSlaves(1), maxSlavesPerHost(0), skipCount(0)
     {
     }
 
@@ -181,16 +186,64 @@
         return ret;
     }
 
+    int activeSlavesCountFor(SimpleJob* job)
+    {
+        int count = 0;
+        QString host = job->url().host();
+
+        if (!host.isEmpty())
+        {
+            QListIterator<Slave *> it (activeSlaves);
+            while (it.hasNext())
+            {
+                if (host == it.next()->host())
+                    count++;
+            }
+
+            QString url = job->url().url();
+            kDebug() << "*** Reserve list count: " << reserveList.count();
+
+            if (reserveList.contains(url)) {
+                kDebug() << "*** Removing paired request for: " << url;
+                reserveList.removeOne(url);
+            } else {
+                count += reserveList.count();
+            }
+        }
+
+        return count;
+    }
+
+    QStringList reserveList;
     QList<SimpleJob *> joblist;
     SlaveList activeSlaves;
     SlaveList idleSlaves;
     CoSlaveMap coSlaves;
     SlaveList coIdleSlaves;
     int maxSlaves;
+    int maxSlavesPerHost;
     int skipCount;
     QString protocol;
 };
 
+static inline bool checkLimits(KIO::SchedulerPrivate::ProtocolInfo *protInfo, SimpleJob *job)
+{
+  const int numActiveSlaves = protInfo->activeSlavesCountFor(job);
+
+#if 0
+    kDebug() << job->url() << ": ";
+    kDebug() << "    protocol :" << job->url().protocol()
+             << ", max :" << protInfo->maxSlaves
+             << ", max host :" << protInfo->maxSlavesPerHost
+             << ", active :" << protInfo->activeSlaves.count()
+             << ", idle :" << protInfo->idleSlaves.count()
+             << ", active for " << job->url().host() << " = " << numActiveSlaves;
+#endif
+
+  return (protInfo->maxSlavesPerHost < 1 || protInfo->maxSlavesPerHost > numActiveSlaves);
+}
+
+
 KIO::SchedulerPrivate::ProtocolInfo *
 KIO::SchedulerPrivate::ProtocolInfoDict::get(const QString &protocol)
 {
@@ -200,6 +253,7 @@
         info = new ProtocolInfo;
         info->protocol = protocol;
         info->maxSlaves = KProtocolInfo::maxSlaves( protocol );
+        info->maxSlavesPerHost = KProtocolInfo::maxSlavesPerHost( protocol );
 
         insert(protocol, info);
     }
@@ -382,9 +436,10 @@
     }
 }
 
-void SchedulerPrivate::doJob(SimpleJob *job) {
+void SchedulerPrivate::doJob(SimpleJob *job) {  
     JobData jobData;
     jobData.protocol = KProtocolManager::slaveProtocol(job->url(), jobData.proxy);
+
 //    kDebug(7006) << "protocol=" << jobData->protocol;
     if (jobCommand(job) == CMD_GET)
     {
@@ -401,15 +456,17 @@
 }
 
 void SchedulerPrivate::scheduleJob(SimpleJob *job) {
+
     newJobs.removeOne(job);
     const JobData& jobData = extraJobData.value(job);
 
     QString protocol = jobData.protocol;
 //    kDebug(7006) << "protocol=" << protocol;
     ProtocolInfo *protInfo = protInfoDict.get(protocol);
-    protInfo->joblist.append(job);
-
-    slaveTimer.start(0);
+    if (!protInfo->joblist.contains(job)) { // scheduleJob already called for this job?
+        protInfo->joblist.append(job);
+        slaveTimer.start(0);
+    }
 }
 
 void SchedulerPrivate::cancelJob(SimpleJob *job) {
@@ -422,6 +479,7 @@
         newJobs.removeAll(job);
         ProtocolInfo *protInfo = protInfoDict.get(jobData.protocol);
         protInfo->joblist.removeAll(job);
+        protInfo->reserveList.removeAll(job->url().url());
 
         // Search all slaves to see if job is in the queue of a coSlave
         foreach(Slave* coSlave, protInfo->allSlaves())
@@ -520,15 +578,15 @@
 
     SimpleJob *job = 0;
     Slave *slave = 0;
-
+    
     if (protInfo->skipCount > 2)
     {
        bool dummy;
        // Prevent starvation. We skip the first entry in the queue at most
        // 2 times in a row. The
        protInfo->skipCount = 0;
-       job = protInfo->joblist.at(0);
-       slave = findIdleSlave(protInfo, job, dummy );
+       job = protInfo->joblist.at(0);     
+       slave = findIdleSlave(protInfo, job, dummy, true);
     }
     else
     {
@@ -538,7 +596,8 @@
        for(int i = 0; (i < protInfo->joblist.count()) && (i < 10); i++)
        {
           job = protInfo->joblist.at(i);
-          slave = findIdleSlave(protInfo, job, exact);
+          slave = findIdleSlave(protInfo, job, exact, true);
+
           if (!firstSlave)
           {
              firstJob = job;
@@ -561,28 +620,37 @@
 
     if (!slave)
     {
-       if ( protInfo->maxSlaves > static_cast<int>(protInfo->activeSlaves.count()) )
-       {
-          newSlave = true;
-          slave = createSlave(protInfo, job, job->url());
-          if (!slave)
-             slaveTimer.start(0);
-       }
+      slave = createSlave(protInfo, job, job->url(), true);
+      if (slave)
+        newSlave = true;
+      else
+        slaveTimer.start(0);
     }
 
     if (!slave)
     {
-//          kDebug(7006) << "No slaves available";
-//          kDebug(7006) << " -- active: " << protInfo->activeSlaves.count();
+       //kDebug() << "No slaves available";
+       //kDebug() << " -- active: " << protInfo->activeSlaves.count();
        return false;
     }
 
+    // Check to make sure the scheduling of this job is dependent on another
+    // job request yet to arrive and if it is add the url of the new job to
+    // to the reserve list. This is done to avoid any potential deadlock
+    // conditions that might occur as a result of scheduling high level jobs,
+    // e.g. cipying file from one ftp server to another one.
+    KUrl url  = pairedRequest(job);
+    if (url.isValid())
+    {
+        kDebug() << "*** PAIRED REQUEST: " << url;
+        protInfoDict.get(url.protocol())->reserveList << url.url();
+    }
+
     protInfo->activeSlaves.append(slave);
     protInfo->idleSlaves.removeAll(slave);
     protInfo->joblist.removeOne(job);
 //        kDebug(7006) << "scheduler: job started " << job;
 
-
     SchedulerPrivate::JobData jobData = extraJobData.value(job);
     setupSlave(slave, job->url(), jobData.protocol, jobData.proxy, newSlave);
     startJob(job, slave);
@@ -635,6 +703,8 @@
 
     foreach( Slave *slave, idleSlaves )
     {
+//       kDebug() << "job protocol: " << protocol << ", slave protocol: " << slave->slaveProtocol();
+//       kDebug() << "job host: " << host << ", slave host: " << slave->host();
        if ((protocol == slave->slaveProtocol()) &&
            (host == slave->host()) &&
            (port == slave->port()) &&
@@ -652,82 +722,100 @@
     return 0;
 }
 
-Slave *SchedulerPrivate::findIdleSlave(ProtocolInfo *protInfo, SimpleJob *job, bool &exact)
+Slave *SchedulerPrivate::findIdleSlave(ProtocolInfo *protInfo, SimpleJob *job,
+                                       bool &exact, bool enforceLimits)
 {
     Slave *slave = 0;
-    JobData jobData = extraJobData.value(job);
 
-    if (jobData.checkOnHold)
+    if (!enforceLimits || checkLimits(protInfo, job))
     {
-       slave = Slave::holdSlave(jobData.protocol, job->url());
-       if (slave)
-          return slave;
+        JobData jobData = extraJobData.value(job);
+
+        if (jobData.checkOnHold)
+        {
+           slave = Slave::holdSlave(jobData.protocol, job->url());
+           if (slave)
+              return slave;
+        }
+
+        if (slaveOnHold)
+        {
+           // Make sure that the job wants to do a GET or a POST, and with no offset
+           bool bCanReuse = (jobCommand(job) == CMD_GET);
+           KIO::TransferJob * tJob = qobject_cast<KIO::TransferJob *>(job);
+           if ( tJob )
+           {
+              bCanReuse = (jobCommand(job) == CMD_GET || jobCommand(job) == CMD_SPECIAL);
+              if ( bCanReuse )
+              {
+                KIO::MetaData outgoing = tJob->outgoingMetaData();
+                QString resume = (!outgoing.contains("resume")) ? QString() : outgoing["resume"];
+                kDebug(7006) << "Resume metadata is" << resume;
+                bCanReuse = (resume.isEmpty() || resume == "0");
+              }
+           }
+//           kDebug(7006) << "bCanReuse = " << bCanReuse;
+           if (bCanReuse)
+           {
+              if (job->url() == urlOnHold)
+              {
+                 kDebug(7006) << "HOLD: Reusing held slave for" << urlOnHold;
+                 slave = slaveOnHold;
+              }
+              else
+              {
+                 kDebug(7006) << "HOLD: Discarding held slave (" << urlOnHold << ")";
+                 slaveOnHold->kill();
+              }
+              slaveOnHold = 0;
+              urlOnHold = KUrl();
+           }
+           if (slave)
+              return slave;
+        }
+
+        slave = searchIdleList(protInfo->idleSlaves, job->url(), jobData.protocol, exact);
     }
-    if (slaveOnHold)
+
+    return slave;
+}
+
+Slave *SchedulerPrivate::createSlave(ProtocolInfo *protInfo, SimpleJob *job,
+                                     const KUrl &url, bool enforceLimits)
+{  
+    Slave *slave = 0;
+    const int slavesCount = protInfo->activeSlaves.count() + protInfo->idleSlaves.count();
+
+    if (!enforceLimits ||
+        (protInfo->maxSlaves > slavesCount && checkLimits(protInfo, job)))
     {
-       // Make sure that the job wants to do a GET or a POST, and with no offset
-       bool bCanReuse = (jobCommand(job) == CMD_GET);
-       KIO::TransferJob * tJob = qobject_cast<KIO::TransferJob *>(job);
-       if ( tJob )
-       {
-          bCanReuse = (jobCommand(job) == CMD_GET || jobCommand(job) == CMD_SPECIAL);
-          if ( bCanReuse )
-          {
-            KIO::MetaData outgoing = tJob->outgoingMetaData();
-            QString resume = (!outgoing.contains("resume")) ? QString() : outgoing["resume"];
-            kDebug(7006) << "Resume metadata is" << resume;
-            bCanReuse = (resume.isEmpty() || resume == "0");
-          }
-       }
-//       kDebug(7006) << "bCanReuse = " << bCanReuse;
-       if (bCanReuse)
-       {
-          if (job->url() == urlOnHold)
-          {
-             kDebug(7006) << "HOLD: Reusing held slave for" << urlOnHold;
-             slave = slaveOnHold;
-          }
-          else
-          {
-             kDebug(7006) << "HOLD: Discarding held slave (" << urlOnHold << ")";
-             slaveOnHold->kill();
-          }
-          slaveOnHold = 0;
-          urlOnHold = KUrl();
-       }
-       if (slave)
-          return slave;
+        int error;
+        QString errortext;
+        slave = Slave::createSlave(protInfo->protocol, url, error, errortext);
+
+        if (slave)
+        {
+            protInfo->idleSlaves.append(slave);
+            q->connect(slave, SIGNAL(slaveDied(KIO::Slave *)),
+                       SLOT(slotSlaveDied(KIO::Slave *)));
+            q->connect(slave, SIGNAL(slaveStatus(pid_t,const QByteArray&,const QString &, bool)),
+                       SLOT(slotSlaveStatus(pid_t,const QByteArray&, const QString &, bool)));
+        }
+        else
+        {
+            kError() << "couldn't create slave:" << errortext;
+            if (job)
+            {
+                protInfo->joblist.removeAll(job);
+                extraJobData.remove(job);
+                job->slotError( error, errortext );
+            }
+        }
     }
 
-    return searchIdleList(protInfo->idleSlaves, job->url(), jobData.protocol, exact);
+    return slave;
 }
 
-Slave *SchedulerPrivate::createSlave(ProtocolInfo *protInfo, SimpleJob *job, const KUrl &url)
-{
-   int error;
-   QString errortext;
-   Slave *slave = Slave::createSlave(protInfo->protocol, url, error, errortext);
-   if (slave)
-   {
-      protInfo->idleSlaves.append(slave);
-      q->connect(slave, SIGNAL(slaveDied(KIO::Slave *)),
-                 SLOT(slotSlaveDied(KIO::Slave *)));
-      q->connect(slave, SIGNAL(slaveStatus(pid_t,const QByteArray&,const QString &, bool)),
-                 SLOT(slotSlaveStatus(pid_t,const QByteArray&, const QString &, bool)));
-   }
-   else
-   {
-      kError() << "couldn't create slave:" << errortext;
-      if (job)
-      {
-         protInfo->joblist.removeAll(job);
-         extraJobData.remove(job);
-         job->slotError( error, errortext );
-      }
-   }
-   return slave;
-}
-
 void SchedulerPrivate::slotSlaveStatus(pid_t, const QByteArray&, const QString &, bool)
 {
 }
@@ -955,16 +1043,15 @@
 {
 //    kDebug(7006) << "_assignJobToSlave( " << job << ", " << slave << ")";
     QString dummy;
-    if ((slave->slaveProtocol() != KProtocolManager::slaveProtocol( job->url(), dummy ))
-        ||
-        (!newJobs.removeAll(job)))
+    ProtocolInfo *protInfo = protInfoDict.get(slave->protocol());
+    if ((slave->slaveProtocol() != KProtocolManager::slaveProtocol( job->url(), dummy )) ||
+        !(protInfo->joblist.removeAll(job) > 0 || newJobs.removeAll(job) > 0))
     {
         kDebug(7006) << "_assignJobToSlave(): ERROR, nonmatching or unknown job.";
         job->kill();
         return false;
     }
 
-    ProtocolInfo *protInfo = protInfoDict.get(slave->protocol());
     JobList *list = protInfo->coSlaves.value(slave);
     assert(list);
     if (!list)
