qmail-send.c

linux下qmail的源码 本人加了一些注释

void job_close(j)
int j;
{
 struct prioq_elt pe;
 struct stat st;

 if (0 < --jo[j].refs) return;

 pe.id = jo[j].id;
 pe.dt = jo[j].retry;
 if (jo[j].flaghiteof && !jo[j].numtodo)
  {
   fnmake_chanaddr(jo[j].id,jo[j].channel);
   if (unlink(fn.s) == -1)
    {
     log3("warning: unable to unlink ",fn.s,"; will try again later\n");
     pe.dt = now() + SLEEP_SYSFAIL;
    }
   else
    {
     int c;
     for (c = 0;c < CHANNELS;++c) if (c != jo[j].channel)
      {
       fnmake_chanaddr(jo[j].id,c);
       if (stat(fn.s,&st) == 0) return; /* more channels going */
       if (errno != error_noent)
	{
         log3("warning: unable to stat ",fn.s,"\n");
	 break; /* this is the only reason for HOPEFULLY */
	}
      }
     pe.dt = now();
     while (!prioq_insert(&pqdone,&pe)) nomem();
     return;
    }
  }

 while (!prioq_insert(&pqchan[jo[j].channel],&pe)) nomem();
}


/* this file is too long ------------------------------------------- BOUNCES */

char *stripvdomprepend(recip)
char *recip;
{
 int i;
 char *domain;
 int domainlen;
 char *prepend;

 i = str_rchr(recip,'@');
 if (!recip[i]) return recip;
 domain = recip + i + 1;
 domainlen = str_len(domain);

 for (i = 0;i <= domainlen;++i)
   if ((i == 0) || (i == domainlen) || (domain[i] == '.'))
     if (prepend = constmap(&mapvdoms,domain + i,domainlen - i))
      {
       if (!*prepend) break;
       i = str_len(prepend);
       if (str_diffn(recip,prepend,i)) break;
       if (recip[i] != '-') break;
       return recip + i + 1;
      }
 return recip;
}

stralloc bouncetext = {0};

void addbounce(id,recip,report)
unsigned long id;
char *recip;
char *report;
{
 int fd;
 int pos;
 int w;
 while (!stralloc_copys(&bouncetext,"<")) nomem();
 while (!stralloc_cats(&bouncetext,stripvdomprepend(recip))) nomem();
 for (pos = 0;pos < bouncetext.len;++pos)
   if (bouncetext.s[pos] == '\n')
     bouncetext.s[pos] = '_';
 while (!stralloc_cats(&bouncetext,">:\n")) nomem();
 while (!stralloc_cats(&bouncetext,report)) nomem();
 if (report[0])
   if (report[str_len(report) - 1] != '\n')
     while (!stralloc_cats(&bouncetext,"\n")) nomem();
 for (pos = bouncetext.len - 2;pos > 0;--pos)
   if (bouncetext.s[pos] == '\n')
     if (bouncetext.s[pos - 1] == '\n')
       bouncetext.s[pos] = '/';
 while (!stralloc_cats(&bouncetext,"\n")) nomem();
 fnmake2_bounce(id);
 for (;;)
  {
   fd = open_append(fn2.s);
   if (fd != -1) break;
   log1("alert: unable to append to bounce message; HELP! sleeping...\n");
   sleep(10);
  }
 pos = 0;
 while (pos < bouncetext.len)
  {
   w = write(fd,bouncetext.s + pos,bouncetext.len - pos);
   if (w <= 0)
    {
     log1("alert: unable to append to bounce message; HELP! sleeping...\n");
     sleep(10);
    }
   else
     pos += w;
  }
 close(fd);
}

int injectbounce(id)
unsigned long id;
{
 struct qmail qqt;
 struct stat st;
 char *bouncesender;
 char *bouncerecip;
 int r;
 int fd;
 substdio ssread;
 char buf[128];
 char inbuf[128];
 static stralloc sender = {0};
 static stralloc quoted = {0};
 datetime_sec birth;
 unsigned long qp;

 if (!getinfo(&sender,&birth,id)) return 0; /* XXX: print warning */

 /* owner-@host-@[] -> owner-@host */
 if (sender.len >= 5)
   if (str_equal(sender.s + sender.len - 5,"-@[]"))
    {
     sender.len -= 4;
     sender.s[sender.len - 1] = 0;
    }

 fnmake2_bounce(id);
 fnmake_mess(id);
 if (stat(fn2.s,&st) == -1)
  {
   if (errno == error_noent)
     return 1;
   log3("warning: unable to stat ",fn2.s,"\n");
   return 0;
  }
 if (str_equal(sender.s,"#@[]"))
   log3("triple bounce: discarding ",fn2.s,"\n");
 else
  {
   if (qmail_open(&qqt) == -1)
    { log1("warning: unable to start qmail-queue, will try later\n"); return 0; }
   qp = qmail_qp(&qqt);

   if (*sender.s) { bouncesender = ""; bouncerecip = sender.s; }
   else { bouncesender = "#@[]"; bouncerecip = doublebounceto.s; }

   while (!newfield_datemake(now())) nomem();
   qmail_put(&qqt,newfield_date.s,newfield_date.len);
   qmail_puts(&qqt,"From: ");
   while (!quote(&quoted,&bouncefrom)) nomem();
   qmail_put(&qqt,quoted.s,quoted.len);
   qmail_puts(&qqt,"@");
   qmail_put(&qqt,bouncehost.s,bouncehost.len);
   qmail_puts(&qqt,"\nTo: ");
   while (!quote2(&quoted,bouncerecip)) nomem();
   qmail_put(&qqt,quoted.s,quoted.len);
   qmail_puts(&qqt,"\n\
Subject: failure notice\n\
\n\
Hi. This is the qmail-send program at ");
   qmail_put(&qqt,bouncehost.s,bouncehost.len);
   qmail_puts(&qqt,*sender.s ? ".\n\
I'm afraid I wasn't able to deliver your message to the following addresses.\n\
This is a permanent error; I've given up. Sorry it didn't work out.\n\
\n\
" : ".\n\
I tried to deliver a bounce message to this address, but the bounce bounced!\n\
\n\
");

   fd = open_read(fn2.s);
   if (fd == -1)
     qmail_fail(&qqt);
   else
    {
     substdio_fdbuf(&ssread,read,fd,inbuf,sizeof(inbuf));
     while ((r = substdio_get(&ssread,buf,sizeof(buf))) > 0)
       qmail_put(&qqt,buf,r);
     close(fd);
     if (r == -1)
       qmail_fail(&qqt);
    }

   qmail_puts(&qqt,*sender.s ? "--- Below this line is a copy of the message.\n\n" : "--- Below this line is the original bounce.\n\n");
   qmail_puts(&qqt,"Return-Path: <");
   while (!quote2(&quoted,sender.s)) nomem();
   qmail_put(&qqt,quoted.s,quoted.len);
   qmail_puts(&qqt,">\n");

   fd = open_read(fn.s);
   if (fd == -1)
     qmail_fail(&qqt);
   else
    {
     substdio_fdbuf(&ssread,read,fd,inbuf,sizeof(inbuf));
     while ((r = substdio_get(&ssread,buf,sizeof(buf))) > 0)
       qmail_put(&qqt,buf,r);
     close(fd);
     if (r == -1)
       qmail_fail(&qqt);
    }

   qmail_from(&qqt,bouncesender);
   qmail_to(&qqt,bouncerecip);
   if (*qmail_close(&qqt))
    { log1("warning: trouble injecting bounce message, will try later\n"); return 0; }

   strnum2[fmt_ulong(strnum2,id)] = 0;
   log2("bounce msg ",strnum2);
   strnum2[fmt_ulong(strnum2,qp)] = 0;
   log3(" qp ",strnum2,"\n");
  }
 if (unlink(fn2.s) == -1)
  {
   log3("warning: unable to unlink ",fn2.s,"\n");
   return 0;
  }
 return 1;
}


/* this file is too long ---------------------------------------- DELIVERIES */

struct del
 {
  int used;
  int j;
  unsigned long delid;
  seek_pos mpos;
  stralloc recip;
 }
;

unsigned long masterdelid = 1;
unsigned int concurrency[CHANNELS] = { 10, 20 };
unsigned int concurrencyused[CHANNELS] = { 0, 0 };
struct del *d[CHANNELS];
stralloc dline[CHANNELS];
char delbuf[2048];

void del_status()
{
  int c;

  log1("status:");
  for (c = 0;c < CHANNELS;++c) {
    strnum2[fmt_ulong(strnum2,(unsigned long) concurrencyused[c])] = 0;
    strnum3[fmt_ulong(strnum3,(unsigned long) concurrency[c])] = 0;
    log2(chanstatusmsg[c],strnum2);
    log2("/",strnum3);
  }
  if (flagexitasap) log1(" exitasap");
  log1("\n");
}

void del_init()
{
 int c;
 int i;
 for (c = 0;c < CHANNELS;++c)
  {
   flagspawnalive[c] = 1;
   while (!(d[c] = (struct del *) alloc(concurrency[c] * sizeof(struct del))))
     nomem();
   for (i = 0;i < concurrency[c];++i)
    { d[c][i].used = 0; d[c][i].recip.s = 0; }
   dline[c].s = 0;
   while (!stralloc_copys(&dline[c],"")) nomem();
  }
 del_status();
}

int del_canexit()
{
 int c;
 for (c = 0;c < CHANNELS;++c)
   if (flagspawnalive[c]) /* if dead, nothing we can do about its jobs */
     if (concurrencyused[c]) return 0;
 return 1;
}

int del_avail(c)
int c;
{
  return flagspawnalive[c] && comm_canwrite(c) && (concurrencyused[c] < concurrency[c]);
}

void del_start(j,mpos,recip)
int j;
seek_pos mpos;
char *recip;
{
 int i;
 int c;

 c = jo[j].channel;
 if (!flagspawnalive[c]) return;
 if (!comm_canwrite(c)) return;

 for (i = 0;i < concurrency[c];++i) if (!d[c][i].used) break;
 if (i == concurrency[c]) return;

 if (!stralloc_copys(&d[c][i].recip,recip)) { nomem(); return; }
 if (!stralloc_0(&d[c][i].recip)) { nomem(); return; }
 d[c][i].j = j; ++jo[j].refs;
 d[c][i].delid = masterdelid++;
 d[c][i].mpos = mpos;
 d[c][i].used = 1; ++concurrencyused[c];

 comm_write(c,i,jo[j].id,jo[j].sender.s,recip);

 strnum2[fmt_ulong(strnum2,d[c][i].delid)] = 0;
 strnum3[fmt_ulong(strnum3,jo[j].id)] = 0;
 log2("starting delivery ",strnum2);
 log3(": msg ",strnum3,tochan[c]);
 logsafe(recip);
 log1("\n");
 del_status();
}

void markdone(c,id,pos)
int c;
unsigned long id;
seek_pos pos;
{
 struct stat st;
 int fd;
 fnmake_chanaddr(id,c);
 for (;;)
  {
   fd = open_write(fn.s);
   if (fd == -1) break;
   if (fstat(fd,&st) == -1) { close(fd); break; }
   if (seek_set(fd,pos) == -1) { close(fd); break; }
   if (write(fd,"D",1) != 1) { close(fd); break; }
   /* further errors -> double delivery without us knowing about it, oh well */
   close(fd);
   return;
  }
 log3("warning: trouble marking ",fn.s,"; message will be delivered twice!\n");
}

void del_dochan(c)
int c;
{
 int r;
 char ch;
 int i;
 int delnum;
 r = read(chanfdin[c],delbuf,sizeof(delbuf));
 if (r == -1) return;
 if (r == 0) { spawndied(c); return; }
 for (i = 0;i < r;++i)
  {
   ch = delbuf[i];
   while (!stralloc_append(&dline[c],&ch)) nomem();
   if (dline[c].len > REPORTMAX)
     dline[c].len = REPORTMAX;
     /* qmail-lspawn and qmail-rspawn are responsible for keeping it short */
     /* but from a security point of view, we don't trust rspawn */
   if (!ch && (dline[c].len > 1))
    {
     delnum = (unsigned int) (unsigned char) dline[c].s[0];
     if ((delnum < 0) || (delnum >= concurrency[c]) || !d[c][delnum].used)
       log1("warning: internal error: delivery report out of range\n");
     else
      {
       strnum3[fmt_ulong(strnum3,d[c][delnum].delid)] = 0;
       if (dline[c].s[1] == 'Z')
	 if (jo[d[c][delnum].j].flagdying)
	  {
	   dline[c].s[1] = 'D';
	   --dline[c].len;
	   while (!stralloc_cats(&dline[c],"I'm not going to try again; this message has been in the queue too long.\n")) nomem();
	   while (!stralloc_0(&dline[c])) nomem();
	  }
       switch(dline[c].s[1])
	{
	 case 'K':
	   log3("delivery ",strnum3,": success: ");
	   logsafe(dline[c].s + 2);
	   log1("\n");
	   markdone(c,jo[d[c][delnum].j].id,d[c][delnum].mpos);
	   --jo[d[c][delnum].j].numtodo;
	   break;
	 case 'Z':
	   log3("delivery ",strnum3,": deferral: ");
	   logsafe(dline[c].s + 2);
	   log1("\n");
	   break;
	 case 'D':
	   log3("delivery ",strnum3,": failure: ");
	   logsafe(dline[c].s + 2);
	   log1("\n");
	   addbounce(jo[d[c][delnum].j].id,d[c][delnum].recip.s,dline[c].s + 2);
	   markdone(c,jo[d[c][delnum].j].id,d[c][delnum].mpos);
	   --jo[d[c][delnum].j].numtodo;
	   break;
	 default:
	   log3("delivery ",strnum3,": report mangled, will defer\n");
	}
       job_close(d[c][delnum].j);
       d[c][delnum].used = 0; --concurrencyused[c];
       del_status();
      }
     dline[c].len = 0;
    }
  }
}

void del_selprep(nfds,rfds)
int *nfds;
fd_set *rfds;
{
 int c;
 for (c = 0;c < CHANNELS;++c)
   if (flagspawnalive[c])
    {
     FD_SET(chanfdin[c],rfds);
     if (*nfds <= chanfdin[c])
       *nfds = chanfdin[c] + 1;
    }
}

void del_do(rfds)
fd_set *rfds;
{
 int c;
 for (c = 0;c < CHANNELS;++c)
   if (flagspawnalive[c])
     if (FD_ISSET(chanfdin[c],rfds))
       del_dochan(c);
}


/* this file is too long -------------------------------------------- PASSES */

struct
 {
  unsigned long id; /* if 0, need a new 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 */
  substdio ss;
  char buf[128];
 }
pass[CHANNELS];

void pass_init()
{
 int c;
 for (c = 0;c < CHANNELS;++c) pass[c].id = 0;
}

void pass_selprep(wakeup)
datetime_sec *wakeup;
{
 int c;
 struct prioq_elt pe;
 if (flagexitasap) return;
 for (c = 0;c < CHANNELS;++c)
   if (pass[c].id)
     if (del_avail(c))
      { *wakeup = 0; return; }
 if (job_avail())
   for (c = 0;c < CHANNELS;++c)
     if (!pass[c].id)
       if (prioq_min(&pqchan[c],&pe))
         if (*wakeup > pe.dt)
           *wakeup = pe.dt;
 if (prioq_min(&pqfail,&pe))
   if (*wakeup > pe.dt)
     *wakeup = pe.dt;
 if (prioq_min(&pqdone,&pe))
   if (*wakeup > pe.dt)
     *wakeup = pe.dt;
}

static datetime_sec squareroot(x) /* result^2 <= x < (result + 1)^2 */
datetime_sec x; /* assuming: >= 0 */
{
 datetime_sec y;
 datetime_sec yy;
 datetime_sec y21;
 int j;

 y = 0; yy = 0;
 for (j = 15;j >= 0;--j)
  {
   y21 = (y << (j + 1)) + (1 << (j + j));
   if (y21 <= x - yy) { y += (1 << j); yy += y21; }
  }
 return y;
}

datetime_sec nextretry(birth,c)
datetime_sec birth;
int c;
{
 int n;

 if (birth > recent) n = 0;
 else n = squareroot(recent - birth); /* no need to add fuzz to recent */
 n += chanskip[c];
 return birth + n * n;
}

void pass_dochan(c)
int c;
{
 datetime_sec birth;
 struct prioq_elt pe;
 static stralloc line = {0};
 int match;

 if (flagexitasap) return;

 if (!pass[c].id)
  {
   if (!job_avail()) return;
   if (!prioq_min(&pqchan[c],&pe)) return;
   if (pe.dt > recent) return;
   fnmake_chanaddr(pe.id,c);

   prioq_delmin(&pqchan[c]);
   pass[c].mpos = 0;
   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; }
   pass[c].id = pe.id;