diff -uPr qmail-1.03-va/prioq.c qmail-1.03-mpl/prioq.c
--- qmail-1.03-va/prioq.c	1998-06-15 12:53:16.000000000 +0200
+++ qmail-1.03-mpl/prioq.c	2004-12-09 09:02:06.000000000 +0100
@@ -4,13 +4,21 @@
 
 GEN_ALLOC_readyplus(prioq,struct prioq_elt,p,len,a,i,n,x,100,prioq_readyplus)
 
-int prioq_insert(pq,pe)
+/* raven+mplimit:
+ * renamed to *_mplimit
+ * added third parameter to set mplpos
+ * added fourth parameter to set numtodo
+ * added prioq_limit() to set mplpos default to 0
+ */
+int prioq_insert_mplimit(pq,pe,mplpos,numtodo)
 prioq *pq;
 struct prioq_elt *pe;
 {
  int i;
  int j;
  if (!prioq_readyplus(pq,1)) return 0;
+ pe->mplpos = mplpos;   /* raven+mplimit */
+ pe->numtodo = numtodo; /* raven+mplimit */
  j = pq->len++;
  while (j)
   {
@@ -23,6 +31,15 @@
  return 1;
 }
 
+inline
+int prioq_insert(pq,pe)
+prioq *pq;
+struct prioq_elt *pe;
+{
+  return prioq_insert_mplimit(pq, pe, 0, 0);
+}
+/* raven-mplimit */
+
 int prioq_min(pq,pe)
 prioq *pq;
 struct prioq_elt *pe;
diff -uPr qmail-1.03-va/prioq.h qmail-1.03-mpl/prioq.h
--- qmail-1.03-va/prioq.h	1998-06-15 12:53:16.000000000 +0200
+++ qmail-1.03-mpl/prioq.h	2004-12-09 08:57:34.000000000 +0100
@@ -3,8 +3,10 @@
 
 #include "datetime.h"
 #include "gen_alloc.h"
+#include "seek.h" /* raven + mplimit */
 
-struct prioq_elt { datetime_sec dt; unsigned long id; } ;
+/* raven + mplimit: added mplpos & numtodo */
+struct prioq_elt { datetime_sec dt; unsigned long id; seek_pos mplpos; int numtodo; };
 
 GEN_ALLOC_typedef(prioq,struct prioq_elt,p,len,a)
 
diff -uPr qmail-1.03-va/qmail-send.c qmail-1.03-mpl/qmail-send.c
--- qmail-1.03-va/qmail-send.c	1998-06-15 12:53:16.000000000 +0200
+++ qmail-1.03-mpl/qmail-send.c	2004-12-09 08:57:11.000000000 +0100
@@ -489,6 +489,7 @@
   int numtodo;
   int flaghiteof;
   int flagdying;
+  seek_pos mplpos; /* raven+mplimit: seek position */
  }
 ;
 
@@ -525,6 +526,8 @@
  jo[j].channel = channel;
  jo[j].numtodo = 0;
  jo[j].flaghiteof = 0;
+ /* raven+mplimit: in pass_dochan() numtodo is restored after job_open,
+  * mplpos is saved before job_close */
  return j;
 }
 
@@ -560,11 +563,16 @@
 	}
       }
      pe.dt = now();
+     /* raven+mplimit: no need to save mplpos & numtodo here */
      while (!prioq_insert(&pqdone,&pe)) nomem();
      return;
     }
   }
 
+ /* raven+mplimit: remember pos/numtodo in message if not eof, else forget */
+ if (! jo[j].flaghiteof)
+ while (!prioq_insert_mplimit(&pqchan[jo[j].channel],&pe,jo[j].mplpos,jo[j].numtodo)) nomem();
+ else
  while (!prioq_insert(&pqchan[jo[j].channel],&pe)) nomem();
 }
 
@@ -782,6 +790,7 @@
 unsigned long masterdelid = 1;
 unsigned int concurrency[CHANNELS] = { 10, 20 };
 unsigned int concurrencyused[CHANNELS] = { 0, 0 };
+unsigned int mplimit[CHANNELS] = { 8, 16 }; /* raven+mplimit: default limits */
 struct del *d[CHANNELS];
 stralloc dline[CHANNELS];
 char delbuf[2048];
@@ -986,6 +995,7 @@
 struct
  {
   unsigned long id; /* if 0, need a new pass */
+  int delivers; /* raven+mplimit: count delivers on this pass */
   int j; /* defined if id; job number */
   int fd; /* defined if id; reading from {local,remote} */
   seek_pos mpos; /* defined if id; mark position */
@@ -1075,9 +1085,15 @@
    pass[c].fd = open_read(fn.s);
    if (pass[c].fd == -1) goto trouble;
    if (!getinfo(&line,&birth,pe.id)) { close(pass[c].fd); goto trouble; }
+   /* raven+mplimit: restore numtodo/stream position (from pe?), clear delivers */
+   pass[c].delivers = 0;
+   pass[c].mpos = (pe.mplpos > 0 ? pe.mplpos : 0); /* good value becouse of changes in prioq.c */
+   seek_set(pass[c].fd, pass[c].mpos);
+   /* raven- */
    pass[c].id = pe.id;
    substdio_fdbuf(&pass[c].ss,read,pass[c].fd,pass[c].buf,sizeof(pass[c].buf));
    pass[c].j = job_open(pe.id,c);
+   jo[pass[c].j].numtodo = (pe.numtodo > 0 ? pe.numtodo : 0); /* raven+mplimit: restore numtodo */
    jo[pass[c].j].retry = nextretry(birth,c);
    jo[pass[c].j].flagdying = (recent > birth + lifetime);
    while (!stralloc_copy(&jo[pass[c].j].sender,&line)) nomem();
@@ -1090,6 +1106,7 @@
    fnmake_chanaddr(pass[c].id,c);
    log3("warning: trouble reading ",fn.s,"; will try again later\n");
    close(pass[c].fd);
+   jo[pass[c].j].mplpos = 0; /* raven+mplimit: clean mplpos (or maybe not ?) */
    job_close(pass[c].j);
    pass[c].id = 0;
    return;
@@ -1107,12 +1124,23 @@
    case 'T':
      ++jo[pass[c].j].numtodo;
      del_start(pass[c].j,pass[c].mpos,line.s + 1);
+     /* raven+mplimit: close job if overlimit */
+     if (++pass[c].delivers > mplimit[c]) {
+      close(pass[c].fd);
+      jo[pass[c].j].mplpos = pass[c].mpos + line.len;   /* remember current position */
+      jo[pass[c].j].retry = recent + (chanskip[c] / 5); /* shorten retry time */
+      job_close(pass[c].j);     /* close current job */
+      pass[c].id = 0;           /* clear this pass */
+      return;
+     };
+     /* raven- */
      break;
    case 'D':
      break;
    default:
      fnmake_chanaddr(pass[c].id,c);
      log3("warning: unknown record type in ",fn.s,"!\n");
+     jo[pass[c].j].mplpos = 0; /* raven+mplimit: clean mplpos */
      close(pass[c].fd);
      job_close(pass[c].j);
      pass[c].id = 0;
@@ -1555,6 +1583,7 @@
    u = (unsigned int) (unsigned char) ch;
    if (concurrency[c] > u) concurrency[c] = u;
    numjobs += concurrency[c];
+   mplimit[c] = concurrency[c] * 4 / 5; /* raven+mplimit: count mplimit */
   }
 
  fnmake_init();
