qmail-send.c

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

   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].retry = nextretry(birth,c);
   jo[pass[c].j].flagdying = (recent > birth + lifetime);
   while (!stralloc_copy(&jo[pass[c].j].sender,&line)) nomem();
  }

 if (!del_avail(c)) return;

 if (getln(&pass[c].ss,&line,&match,'\0') == -1)
  {
   fnmake_chanaddr(pass[c].id,c);
   log3("warning: trouble reading ",fn.s,"; will try again later\n");
   close(pass[c].fd);
   job_close(pass[c].j);
   pass[c].id = 0;
   return;
  }
 if (!match)
  {
   close(pass[c].fd);
   jo[pass[c].j].flaghiteof = 1;
   job_close(pass[c].j);
   pass[c].id = 0;
   return;
  }
 switch(line.s[0])
  {
   case 'T':
     ++jo[pass[c].j].numtodo;
     del_start(pass[c].j,pass[c].mpos,line.s + 1);
     break;
   case 'D':
     break;
   default:
     fnmake_chanaddr(pass[c].id,c);
     log3("warning: unknown record type in ",fn.s,"!\n");
     close(pass[c].fd);
     job_close(pass[c].j);
     pass[c].id = 0;
     return;
  }

 pass[c].mpos += line.len;
 return;

 trouble:
 log3("warning: trouble opening ",fn.s,"; will try again later\n");
 pe.dt = recent + SLEEP_SYSFAIL;
 while (!prioq_insert(&pqchan[c],&pe)) nomem();
}

void messdone(id)
unsigned long id;
{
 char ch;
 int c;
 struct prioq_elt pe;
 struct stat st;

 for (c = 0;c < CHANNELS;++c)
  {
   fnmake_chanaddr(id,c);
   if (stat(fn.s,&st) == 0) return; /* false alarm; consequence of HOPEFULLY */
   if (errno != error_noent)
    {
     log3("warning: unable to stat ",fn.s,"; will try again later\n");
     goto fail;
    }
  }

 fnmake_todo(id);
 if (stat(fn.s,&st) == 0) return;
 if (errno != error_noent)
  {
   log3("warning: unable to stat ",fn.s,"; will try again later\n");
   goto fail;
  }
 
 fnmake_info(id);
 if (stat(fn.s,&st) == -1)
  {
   if (errno == error_noent) return;
   log3("warning: unable to stat ",fn.s,"; will try again later\n");
   goto fail;
  }
 
 /* -todo +info -local -remote ?bounce */
 if (!injectbounce(id))
   goto fail; /* injectbounce() produced error message */

 strnum3[fmt_ulong(strnum3,id)] = 0;
 log3("end msg ",strnum3,"\n");

 /* -todo +info -local -remote -bounce */
 fnmake_info(id);
 if (unlink(fn.s) == -1)
  {
   log3("warning: unable to unlink ",fn.s,"; will try again later\n");
   goto fail;
  }

 /* -todo -info -local -remote -bounce; we can relax */
 fnmake_foop(id);
 if (substdio_putflush(&sstoqc,fn.s,fn.len) == -1) { cleandied(); return; }
 if (substdio_get(&ssfromqc,&ch,1) != 1) { cleandied(); return; }
 if (ch != '+')
   log3("warning: qmail-clean unable to clean up ",fn.s,"\n");

 return;

 fail:
 pe.id = id; pe.dt = now() + SLEEP_SYSFAIL;
 while (!prioq_insert(&pqdone,&pe)) nomem();
}

void pass_do()
{
 int c;
 struct prioq_elt pe;

 for (c = 0;c < CHANNELS;++c) pass_dochan(c);
 if (prioq_min(&pqfail,&pe))
   if (pe.dt <= recent)
    {
     prioq_delmin(&pqfail);
     pqadd(pe.id);
    }
 if (prioq_min(&pqdone,&pe))
   if (pe.dt <= recent)
    {
     prioq_delmin(&pqdone);
     messdone(pe.id);
    }
}


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

datetime_sec nexttodorun;
DIR *tododir; /* if 0, have to opendir again */
stralloc todoline = {0};
char todobuf[SUBSTDIO_INSIZE];
char todobufinfo[512];
char todobufchan[CHANNELS][1024];

void todo_init()
{
 tododir = 0;
 nexttodorun = now();
 trigger_set();
}

void todo_selprep(nfds,rfds,wakeup)
int *nfds;
fd_set *rfds;
datetime_sec *wakeup;
{
 if (flagexitasap) return;
 trigger_selprep(nfds,rfds);
 if (tododir) *wakeup = 0;
 if (*wakeup > nexttodorun) *wakeup = nexttodorun;
}

void todo_do(rfds)
fd_set *rfds;
{
 struct stat st;
 substdio ss; int fd;
 substdio ssinfo; int fdinfo;
 substdio sschan[CHANNELS];
 int fdchan[CHANNELS];
 int flagchan[CHANNELS];
 struct prioq_elt pe;
 char ch;
 int match;
 unsigned long id;
 unsigned int len;
 direntry *d;
 int c;
 unsigned long uid;
 unsigned long pid;

 fd = -1;
 fdinfo = -1;
 for (c = 0;c < CHANNELS;++c) fdchan[c] = -1;

 if (flagexitasap) return;

 if (!tododir)
  {
   if (!trigger_pulled(rfds))
     if (recent < nexttodorun)
       return;
   trigger_set();
   tododir = opendir("todo");
   if (!tododir)
    {
     pausedir("todo");
     return;
    }
   nexttodorun = recent + SLEEP_TODO;
  }

 d = readdir(tododir);
 if (!d)
  {
   closedir(tododir);
   tododir = 0;
   return;
  }
 if (str_equal(d->d_name,".")) return;
 if (str_equal(d->d_name,"..")) return;
 len = scan_ulong(d->d_name,&id);
 if (!len || d->d_name[len]) return;

 fnmake_todo(id);

 fd = open_read(fn.s);
 if (fd == -1) { log3("warning: unable to open ",fn.s,"\n"); return; }

 fnmake_mess(id);
 /* just for the statistics */
 if (stat(fn.s,&st) == -1)
  { log3("warning: unable to stat ",fn.s,"\n"); goto fail; }

 for (c = 0;c < CHANNELS;++c)
  {
   fnmake_chanaddr(id,c);
   if (unlink(fn.s) == -1) if (errno != error_noent)
    { log3("warning: unable to unlink ",fn.s,"\n"); goto fail; }
  }

 fnmake_info(id);
 if (unlink(fn.s) == -1) if (errno != error_noent)
  { log3("warning: unable to unlink ",fn.s,"\n"); goto fail; }

 fdinfo = open_excl(fn.s);
 if (fdinfo == -1)
  { log3("warning: unable to create ",fn.s,"\n"); goto fail; }

 strnum3[fmt_ulong(strnum3,id)] = 0;
 log3("new msg ",strnum3,"\n");

 for (c = 0;c < CHANNELS;++c) flagchan[c] = 0;

 substdio_fdbuf(&ss,read,fd,todobuf,sizeof(todobuf));
 substdio_fdbuf(&ssinfo,write,fdinfo,todobufinfo,sizeof(todobufinfo));

 uid = 0;
 pid = 0;

 for (;;)
  {
   if (getln(&ss,&todoline,&match,'\0') == -1)
    {
     /* perhaps we're out of memory, perhaps an I/O error */
     fnmake_todo(id);
     log3("warning: trouble reading ",fn.s,"\n"); goto fail;
    }
   if (!match) break;

   switch(todoline.s[0])
    {
     case 'u':
       scan_ulong(todoline.s + 1,&uid);
       break;
     case 'p':
       scan_ulong(todoline.s + 1,&pid);
       break;
     case 'F':
       if (substdio_putflush(&ssinfo,todoline.s,todoline.len) == -1)
	{
	 fnmake_info(id);
         log3("warning: trouble writing to ",fn.s,"\n"); goto fail;
	}
       log2("info msg ",strnum3);
       strnum2[fmt_ulong(strnum2,(unsigned long) st.st_size)] = 0;
       log2(": bytes ",strnum2);
       log1(" from <"); logsafe(todoline.s + 1);
       strnum2[fmt_ulong(strnum2,pid)] = 0;
       log2("> qp ",strnum2);
       strnum2[fmt_ulong(strnum2,uid)] = 0;
       log2(" uid ",strnum2);
       log1("\n");
       break;
     case 'T':
       switch(rewrite(todoline.s + 1))
	{
	 case 0: nomem(); goto fail;
	 case 2: c = 1; break;
	 default: c = 0; break;
        }
       if (fdchan[c] == -1)
	{
	 fnmake_chanaddr(id,c);
	 fdchan[c] = open_excl(fn.s);
	 if (fdchan[c] == -1)
          { log3("warning: unable to create ",fn.s,"\n"); goto fail; }
	 substdio_fdbuf(&sschan[c]
	   ,write,fdchan[c],todobufchan[c],sizeof(todobufchan[c]));
	 flagchan[c] = 1;
	}
       if (substdio_bput(&sschan[c],rwline.s,rwline.len) == -1)
        {
	 fnmake_chanaddr(id,c);
         log3("warning: trouble writing to ",fn.s,"\n"); goto fail;
        }
       break;
     default:
       fnmake_todo(id);
       log3("warning: unknown record type in ",fn.s,"\n"); goto fail;
    }
  }

 close(fd); fd = -1;

 fnmake_info(id);
 if (substdio_flush(&ssinfo) == -1)
  { log3("warning: trouble writing to ",fn.s,"\n"); goto fail; }
 if (fsync(fdinfo) == -1)
  { log3("warning: trouble fsyncing ",fn.s,"\n"); goto fail; }
 close(fdinfo); fdinfo = -1;

 for (c = 0;c < CHANNELS;++c)
   if (fdchan[c] != -1)
    {
     fnmake_chanaddr(id,c);
     if (substdio_flush(&sschan[c]) == -1)
      { log3("warning: trouble writing to ",fn.s,"\n"); goto fail; }
     if (fsync(fdchan[c]) == -1)
      { log3("warning: trouble fsyncing ",fn.s,"\n"); goto fail; }
     close(fdchan[c]); fdchan[c] = -1;
    }

 fnmake_todo(id);
 if (substdio_putflush(&sstoqc,fn.s,fn.len) == -1) { cleandied(); return; }
 if (substdio_get(&ssfromqc,&ch,1) != 1) { cleandied(); return; }
 if (ch != '+')
  {
   log3("warning: qmail-clean unable to clean up ",fn.s,"\n");
   return;
  }

 pe.id = id; pe.dt = now();
 for (c = 0;c < CHANNELS;++c)
   if (flagchan[c])
     while (!prioq_insert(&pqchan[c],&pe)) nomem();

 for (c = 0;c < CHANNELS;++c) if (flagchan[c]) break;
 if (c == CHANNELS)
   while (!prioq_insert(&pqdone,&pe)) nomem();

 return;

 fail:
 if (fd != -1) close(fd);
 if (fdinfo != -1) close(fdinfo);
 for (c = 0;c < CHANNELS;++c)
   if (fdchan[c] != -1) close(fdchan[c]);
}


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

int getcontrols() { if (control_init() == -1) return 0;

 //读入queuelifetime，queuelifetime用来指定一个邮件在队列中的最大存活时间，默认是
 //604800秒，当一个邮件达到这个时间线后，qmail将最后一次尝试发送这封邮件，如果依然
 //失败，会将其从队列中删除
 if (control_readint(&lifetime,"control/queuelifetime") == -1) return 0;

 //读入concurrencylocal，concurrencylocal用来决定qmail可以同时运行的本地投递进程的数目，默认值是10，
 //标准安装的qmail最大可设值是120.安装的时候可以通过修改conf-spawn来将其变为255以下
 //任何数目。
 if (control_readint(&concurrency[0],"control/concurrencylocal") == -1) return 0;

 //读入concurrencyremote，concurrencyremote用来决定可以同时运行的远程投递进程的数目，
 //默认值是20，就是说，qmail可以在同一时间内同时发送20封邮件。
 if (control_readint(&concurrency[1],"control/concurrencyremote") == -1) return 0;
 if (control_rldef(&envnoathost,"control/envnoathost",1,"envnoathost") != 1) return 0;
 if (control_rldef(&bouncefrom,"control/bouncefrom",0,"MAILER-DAEMON") != 1) return 0;
 if (control_rldef(&bouncehost,"control/bouncehost",1,"bouncehost") != 1) return 0;
 if (control_rldef(&doublebouncehost,"control/doublebouncehost",1,"doublebouncehost") != 1) return 0;
 if (control_rldef(&doublebounceto,"control/doublebounceto",0,"postmaster") != 1) return 0;
 if (!stralloc_cats(&doublebounceto,"@")) return 0;
 if (!stralloc_cat(&doublebounceto,&doublebouncehost)) return 0;
 if (!stralloc_0(&doublebounceto)) return 0;
 if (control_readfile(&locals,"control/locals",1) != 1) return 0;
 if (!constmap_init(&maplocals,locals.s,locals.len,0)) return 0;
 switch(control_readfile(&percenthack,"control/percenthack",0))
  {
   case -1: return 0;
   case 0: if (!constmap_init(&mappercenthack,"",0,0)) return 0; break;
   case 1: if (!constmap_init(&mappercenthack,percenthack.s,percenthack.len,0)) return 0; break;
  }
 //读入virtualdomains，virtualdomains让qmail接受本地邮件以外，还接受virtualdomains
 //指定的域或者邮件地址，就是常说的虚拟域支持
 switch(control_readfile(&vdoms,"control/virtualdomains",0))
  {
   case -1: return 0;
   case 0: if (!constmap_init(&mapvdoms,"",0,1)) return 0; break;
   case 1: if (!constmap_init(&mapvdoms,vdoms.s,vdoms.len,1)) return 0; break;
  }
 return 1; }

stralloc newlocals = {0};
stralloc newvdoms = {0};

void regetcontrols()
{
 int r;

 if (control_readfile(&newlocals,"control/locals",1) != 1)
  { log1("alert: unable to reread control/locals\n"); return; }
 r = control_readfile(&newvdoms,"control/virtualdomains",0);
 if (r == -1)
  { log1("alert: unable to reread control/virtualdomains\n"); return; }

 constmap_free(&maplocals);
 constmap_free(&mapvdoms);

 while (!stralloc_copy(&locals,&newlocals)) nomem();
 while (!constmap_init(&maplocals,locals.s,locals.len,0)) nomem();

 if (r)
  {
   while (!stralloc_copy(&vdoms,&newvdoms)) nomem();
   while (!constmap_init(&mapvdoms,vdoms.s,vdoms.len,1)) nomem();
  }
 else
   while (!constmap_init(&mapvdoms,"",0,1)) nomem();
}

void reread()
{
 if (chdir(auto_qmail) == -1)
  {
   log1("alert: unable to reread controls: unable to switch to home directory\n");
   return;
  }
 regetcontrols();
 while (chdir("queue") == -1)
  {
   log1("alert: unable to switch back to queue directory; HELP! sleeping...\n");
   sleep(10);
  }
}

void main()
{
 int fd;
 datetime_sec wakeup;
 fd_set rfds;
 fd_set wfds;
 int nfds;
 struct timeval tv;
 int c;

 if (chdir(auto_qmail) == -1)
  { log1("alert: cannot start: unable to switch to home directory\n"); _exit(111); }
 if (!getcontrols())
  { log1("alert: cannot start: unable to read controls\n"); _exit(111); }
 if (chdir("queue") == -1)
  { log1("alert: cannot start: unable to switch to queue directory\n"); _exit(111); }
 sig_pipeignore();
 sig_termcatch(sigterm);
 sig_alarmcatch(sigalrm);
 sig_hangupcatch(sighup);
 sig_childdefault();
 umask(077);

 fd = open_write("lock/sendmutex");
 if (fd == -1)
  { log1("alert: cannot start: unable to open mutex\n"); _exit(111); }
 if (lock_exnb(fd) == -1)
  { log1("alert: cannot start: qmail-send is already running\n"); _exit(111); }

 numjobs = 0;
 for (c = 0;c < CHANNELS;++c)
  {
   char ch;
   int u;
   int r;
   do
     r = read(chanfdin[c],&ch,1);
   while ((r == -1) && (errno == error_intr));
   if (r < 1)
    { log1("alert: cannot start: hath the daemon spawn no fire?\n"); _exit(111); }
   u = (unsigned int) (unsigned char) ch;
   if (concurrency[c] > u) concurrency[c] = u;
   numjobs += concurrency[c];
  }

 fnmake_init();

 comm_init();

 pqstart();
 job_init();
 del_init();
 pass_init();
 todo_init();
 cleanup_init();

 while (!flagexitasap || !del_canexit())
  {
   recent = now();

   if (flagrunasap) { flagrunasap = 0; pqrun(); }
   if (flagreadasap) { flagreadasap = 0; reread(); }

   wakeup = recent + SLEEP_FOREVER;
   FD_ZERO(&rfds);
   FD_ZERO(&wfds);
   nfds = 1;

   comm_selprep(&nfds,&wfds);
   del_selprep(&nfds,&rfds);
   pass_selprep(&wakeup);
   todo_selprep(&nfds,&rfds,&wakeup);
   cleanup_selprep(&wakeup);

   if (wakeup <= recent) tv.tv_sec = 0;
   else tv.tv_sec = wakeup - recent + SLEEP_FUZZ;
   tv.tv_usec = 0;

   if (select(nfds,&rfds,&wfds,(fd_set *) 0,&tv) == -1)
     if (errno == error_intr)
       ;
     else
       log1("warning: trouble in select\n");
   else
    {
     recent = now();

     comm_do(&wfds);
     del_do(&rfds);
     todo_do(&rfds);
     pass_do();
     cleanup_do();
    }
  }
 pqfinish();
 log1("status: exiting\n");
 _exit(0);
}