dns-te~1.cc

一百个病毒的源代码 包括熊猫烧香等 极其具有研究价值

    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.
void
read_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.
void
store_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);
}

void
usage(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);
}