for (from = after; from != end;)      *to++ = *from++;    lp->lenafter = after_len + 1;    if (before_len) {      lp->logbefore = ++to;      end = before + before_len;	// Sentinel again.      for (from = before; from != end;)	*to++ = *from++;    }    lp->lenbefore = before_len;  }  return lp;}// Read DB entries created by earlier runs into reslist,// marked as 'not new' (false).// If skipempty is true, do not read in negative cache entries.voidread_db(DatedStringDb *table, BoolStringMap &reslist, bool skipempty){  DatedStringDbCursor *iterator = new DatedStringDbCursor(table);  char *key, *data;  time_t when;  int inserted = 0;  while (iterator->get(&key, &data, DB_NEXT, &when) == 0)    {      if (*data || !skipempty) {	BoolString k(key, false), v(data, false);	reslist[k] = v;	++inserted;      }    }  iterator->close();  if (verbose)    fprintf(stderr, "read %d addresses from DB file\n", inserted);}// Write out new entries to the DB file.voidstore_db(DatedStringDb *table, BoolStringMap &reslist){  BoolStringMap::iterator it;  char *key, *data;  time_t when;  int added = 0;  time(&when);  for (it = reslist.begin(); it != reslist.end(); it++) {    BoolString ipaddr = (*it).first, value = (*it).second;    if (value.get_flag()) {      if (verbose > 1)	fprintf(stderr, "%s new address\n", ipaddr.get_str());      try	{	  table->put(ipaddr.get_str(), value.get_str(), &when);	}      catch (DbException &dbe)	{	  fprintf(stderr, "DB error storing (%s,%s): %s\n",		  ipaddr.get_str(), value.get_str(), dbe.what());	}      ++added;      value.set_flag(false);    }  }  table->sync();  if (verbose)    fprintf(stderr, "added %d addresses to DB file\n", added);}voidusage(char *prog){  fprintf(stderr,	  "Usage: %s [-v...] [-orsz] [-d db-file] [-c adns-conf-str] [-m mark-size] [-p parallel-queries] [-f skip-fields]\n",	  prog);  exit(1);}int main(int argc, char *const *argv){  adns_state ads;  int r;  adns_initflags aflags;  int outstanding = 0;  size_t qsize;  gzFile zout = NULL;  adns_query qu;  adns_answer *ans;  DatedStringDb *resolved;  BoolStringMap reslist;  LogEntryQue lq;  char *dbhome = NULL;  char *dbfile = DEFAULT_DBFILE;  char *adnsconf = NULL;  long marksize = 0, lastmark = -1;  bool syncmark = false;  bool copylines = false;  bool reresolve = false;  int skip_fields = 0;  int parallel_queries = DEFAULT_PARALLEL_QUERIES;  QueryStats stats;  LogEntry *lp;  char *dom;  program_name = argv[0];  while ((r = getopt(argc, argv, "c:d:f:m:op:rsvz")) != -1) {    switch (r) {    case 'c':      adnsconf = optarg;      break;    case 'd':      dbfile = optarg;      break;    case 'f':      skip_fields = atoi(optarg);      break;    case 'm':      marksize = atol(optarg);      break;    case 'o':      copylines = true;      break;    case 'p':      parallel_queries = atoi(optarg);      break;    case 'r':      reresolve = true;      break;    case 's':      syncmark = true;      break;    case 'v':      verbose++;      break;    case 'z':      zout = gzdopen(1, "wb");      if (!zout) {	fprintf(stderr, "%s: fatal error: %s: %s\n",		program_name, "gzdopen", strerror(errno));	exit(1);      }      break;    default:      usage(argv[0]);      break;    }  }  // We get back here when terminating abnormally.  r = setjmp(getback);  if (r == 1)    goto out;  else if (r == 2) {    if (copylines) {      if (zout) {	if (lp->lenbefore)	  gzwrite(zout, lp->logbefore, lp->lenbefore);	gzputs(zout, lp->ipaddr);	gzwrite(zout, lp->logafter, lp->lenafter);      } else {	if (lp->lenbefore)	  fwrite(lp->logbefore, lp->lenbefore, 1, stdout);	fputs(lp->ipaddr, stdout);	fwrite(lp->logafter, lp->lenafter, 1, stdout);      }    }    goto timed_out;  }  if (verbose > 2)    aflags = (adns_initflags) adns_if_debug;  else    aflags = (adns_initflags) 0;  if (adnsconf)    r = adns_init_strcfg(&ads, aflags, stderr, adnsconf);  else    r = adns_init(&ads, aflags, stderr);  if (r)    fatal_errno("adns_init", r);  try    {      resolved = new DatedStringDb(dbhome, *dbfile ? dbfile : NULL);      set_handlers();      if (*dbfile)	read_db(resolved, reslist, reresolve);      else	stats.cached = -1;    }  catch (DbException &dbe)    {      fprintf(stderr, "DB error opening %s: %s\n",	      *dbfile ? dbfile : "memory DB", dbe.what());      exit(1);    }  if (copylines)    parallel_queries *= COPYLINES_MULTIPLIER;  while ((lp = read_ipaddr(stdin, copylines, skip_fields))) {    ++stats.linesread;    if (marksize && stats.linesread % marksize == 0) {      if (copylines)	fprintf(stderr, "On line %ld, %d queries outstanding, %d lines buffered\n",		stats.linesread, outstanding, lq.size());      else	fprintf(stderr, "On line %ld, %d queries outstanding\n",		stats.linesread, outstanding);      if (syncmark && *dbfile)	store_db(resolved, reslist);    }    if (reopen && *dbfile) {      if (verbose) {	fprintf(stderr, "%s: received hangup signal; reopening DB file\n", program_name);      }      try	{	  delete resolved;	  resolved = new DatedStringDb(dbhome, dbfile);	}      catch (DbException &dbe)	{	  fprintf(stderr, "DB error reopening %s: %s\n",		  dbfile, dbe.what());	  exit(1);	}      store_db(resolved, reslist);      reopen = 0;    }    // When doing copylines:    // We need to save the log files for all entries,    // not just those that we submit.  The log line must be    // in the deque.    //    // When we read a log file entry to enqueue it, the following    // situations are possible:    // 0. The IP address field is syntactically invalid.    //    There is no answer cached in the map.    //    We don't need a query ID (assume it's 0).    // 1. The answer is cached in the map, and is not "?".    //    We don't need a query ID (assume it's 0).    // 2. The answer is cached in the map, and is "?", meaning    //    that the query is in process for some earlier entry.    //    By the time we dequeue this entry, the earlier entry will    //    have been written out, so the answer will be in the map.    //    We don't need a query ID (assume it's 0).    // 3. The answer is not in the map.  We need to submit a    //    new query for it and save the query ID.    //    // When we dequeue a log file entry, the following    // situations are possible:    // 0. The query ID is 0 and there's no answer in the map,    //    so the IP address is syntactically invalid.    //    We can write out the line.    // 1. The query ID is 0.  We know the answer is in the map,    //    and is not "?".  We can write out the line.    // 2. The query ID is not 0.  The query is in process.    //    We need to wait on the query ID for it.    switch (submit_query(ads, reslist, lp)) {    case sb_invalid:      ++stats.invalid;      break;    case sb_pending:      break;    case sb_known:      break;    case sb_cached:      ++stats.cached;      break;    case sb_submitted:      ++outstanding;      ++stats.submitted;      break;    }    if (copylines)      lq.push_back(lp);    // The goal is to keep the query pipeline full, so only pick off    // one answer.    if (copylines)      assert(lq.size() <= parallel_queries);    else      assert(outstanding <= parallel_queries);    if (copylines ? (lq.size() == parallel_queries) : (outstanding == parallel_queries)) {      if (copylines) {	lp = lq[0];	lq.pop_front();	qu = lp->qu;      } else {	qu = 0;      }      if (qu || !copylines) {	// It would be best if the answer were always ready here, so the	// adns_wait call wouldn't block.  That would happen if the time	// it takes to generate a full pipeline of queries were equal to	// (or greater than, though that wouldn't help) the maximum time it	// takes to get back a response (or timeout).  Otherwise, the	// pipeline will stall.	//	// For example, with the default (-p 1000) for -o of 20,000 lines	// buffered, there may be about 800 queries outstanding.	// On a Pentium III/500, we can generate about 1400 queries per second.	// If we set udpmaxretries:4 for an 8 second timeout, then we	// may generate no more than 800/8=100 queries per second to	// avoid all pipeline stalls on timeouts.  8*1400=11200; round to 12000.	// -p 12000 will give better performance assuming you have enough RAM.	//	// Even if you get the main loop optimal, the pipeline will	// probably stall during the drain time at the end.	r = adns_wait(ads, &qu, &ans, (void **) &lp);	if (r)	  fatal_errno("adns_wait", r);	dom = process_answer(ans, lp->ipaddr, reslist);	if (*dom)	  ++stats.successful;	--outstanding;      } else {	// It's already in the list.	BoolString key(lp->ipaddr, false), value;	BoolStringMap::iterator	it = reslist.find(key);	if (it != reslist.end()) {	  value = (*it).second;	  dom = value.get_str();	} else {	  // For a syntactically invalid IP address, there's no answer.	  // Print it unchanged.	  dom = lp->ipaddr;	}	ans = NULL;      }      if (copylines) {	if (zout) {	  if (lp->lenbefore)	    gzwrite(zout, lp->logbefore, lp->lenbefore);	  gzputs(zout, *dom ? dom : lp->ipaddr);	  gzwrite(zout, lp->logafter, lp->lenafter);	} else {	  if (lp->lenbefore)	    fwrite(lp->logbefore, lp->lenbefore, 1, stdout);	  fputs(*dom ? dom : lp->ipaddr, stdout);	  fwrite(lp->logafter, lp->lenafter, 1, stdout);	}      }      if (ans)	free(ans);      free(lp);    }  }  if (verbose || marksize) {    fprintf(stderr, "Read last line %ld, %d queries outstanding\n",	    stats.linesread, outstanding);    if (syncmark && *dbfile)      store_db(resolved, reslist);  }  // Pick up the stragglers; let the pipeline drain.  if (marksize) {    marksize = outstanding / 10;    if (!marksize)      marksize = 1;  }  while (copylines ? ((qsize = lq.size()) > 0) : (outstanding > 0)) {    if (marksize && outstanding % marksize == 0 && outstanding != lastmark) {      lastmark = outstanding;      if (copylines)	fprintf(stderr, "%d queries outstanding, %d lines buffered\n",		outstanding, (int) qsize);      else	fprintf(stderr, "%d queries outstanding\n", outstanding);    }    // Yes, the body of this loop is identical to the inner loop above.    // I can't think of an easy way to roll them together, given all    // the variables they use.    if (copylines) {      lp = lq[0];      lq.pop_front();      qu = lp->qu;    } else {      qu = 0;    }    if (qu || !copylines) {      r = adns_wait(ads, &qu, &ans, (void **) &lp);      if (r)	fatal_errno("adns_wait", r);      dom = process_answer(ans, lp->ipaddr, reslist);      if (*dom)	++stats.successful;      --outstanding;    } else {      // It's already in the list.      BoolString key(lp->ipaddr, false), value;      BoolStringMap::iterator it = reslist.find(key);      if (it != reslist.end()) {	value = (*it).second;	dom = value.get_str();      } else {	dom = lp->ipaddr;      }      ans = NULL;    }    if (copylines) {      if (zout) {	if (lp->lenbefore)	  gzwrite(zout, lp->logbefore, lp->lenbefore);	gzputs(zout, *dom ? dom : lp->ipaddr);	gzwrite(zout, lp->logafter, lp->lenafter);      } else {	if (lp->lenbefore)	  fwrite(lp->logbefore, lp->lenbefore, 1, stdout);	fputs(*dom ? dom : lp->ipaddr, stdout);	fwrite(lp->logafter, lp->lenafter, 1, stdout);      }    }    if (ans)      free(ans);    free(lp);  } timed_out:  if (*dbfile)    store_db(resolved, reslist); out:  if (zout)    gzclose(zout);  try    {      resolved->close();    }  catch (DbException &dbe)    {      fprintf(stderr, "DB error closing %s: %s\n",	      *dbfile ? dbfile : "memory DB", dbe.what());    }  adns_finish(ads);  stats.print();  exit (0);}