chiark - git - mdw - tripe-android/blob - peers.scala

   1 package uk.org.distorted.tripe; package object peers {
   2
   3 import java.io.{BufferedReader, File, FileReader, Reader};
   4 import java.net.{InetAddress, Inet4Address, Inet6Address,
   5                  UnknownHostException};
   6
   7 import scala.collection.mutable.{HashMap, HashSet};
   8 import scala.concurrent.Channel;
   9 import scala.util.control.Breaks;
  10 import scala.util.matching.Regex;
  11
  12 val RX_COMMENT = """(?x) ^ \s* (?: [;\#] .* )? $""".r;
  13 val RX_GRPHDR = """(?x) ^ \s* \[ (.*) \] \s* $""".r;
  14 val RX_ASSGN = """(?x) ^
  15         ([^\s:=] (?: [^:=]* [^\s:=])?)
  16         \s* [:=] \s*
  17         (| [^\s\#;]\S* (?: \s+ [^\s\#;]\S*)*)
  18         (?: \s+ (?: [;\#].*)? )? $""".r;
  19 val RX_CONT = """(?x) ^ \s+
  20         (| [^\s\#;]\S* (?: \s+ [^\s\#;]\S*)*)
  21         (?: \s+ (?: [;\#].*)? )? $""".r;
  22 val RX_REF = """(?x) \$ \( ([^)]+) \)""".r;
  23 val RX_RESOLVE = """(?x) \$ ([46*]*) \[ ([^\]]+) \]""".r;
  24 val RX_PARENT = """(?x) [^\s,]+""".r
  25
  26 object BulkResolver {
  27   val BREAK = new Breaks;
  28 }
  29
  30 class BulkResolver(val nthreads: Int = 8) {
  31   import BulkResolver.BREAK._;
  32   class Host(val name: String) {
  33     var a4, a6: Seq[InetAddress] = Seq.empty;
  34
  35     def addaddr(a: InetAddress) { a match {
  36       case _: Inet4Address => a4 +:= a;
  37       case _: Inet6Address => a6 +:= a;
  38       case _ => ();
  39     } }
  40
  41     def get(flags: String): Seq[InetAddress] = {
  42       var wanta4, wanta6, any, all = false;
  43       var b = Seq.newBuilder[InetAddress];
  44       for (ch <- flags) ch match {
  45         case '*' => all = true;
  46         case '4' => wanta4 = true; any = true;
  47         case '6' => wanta6 = true; any = true;
  48         case _ => ???
  49       }
  50       if (!any) { wanta4 = true; wanta6 = true; }
  51       if (wanta4) b ++= a4;
  52       if (wanta6) b ++= a6;
  53       (all, b.result) match {
  54         case (true, aa) => aa
  55         case (false, aa@(Nil | Seq(_))) => aa
  56         case (false, Seq(a, _*)) => Seq(a)
  57       }
  58     }
  59   }
  60   val ch = new Channel[Host];
  61   val map = HashMap[String, Host]();
  62   var preparing = true;
  63
  64   val workers = Array.tabulate(nthreads) { i =>
  65     thread(s"resolver worker #$i") {
  66       loopUnit { exit =>
  67         val host = ch.read; if (host == null) exit;
  68 println(s";; ${Thread.currentThread.getName} resolving `${host.name}'");
  69         try {
  70           for (a <- InetAddress.getAllByName(host.name)) host.addaddr(a);
  71         } catch { case e: UnknownHostException => () }
  72       }
  73 println(s";; ${Thread.currentThread.getName} done'");
  74       ch.write(null);
  75     }
  76   }
  77
  78   def prepare(name: String) {
  79 println(s";; prepare host `$name'");
  80     assert(preparing);
  81     if (!(map contains name)) {
  82       val host = new Host(name);
  83       map(name) = host;
  84       ch.write(host);
  85     }
  86   }
  87
  88   def finish() {
  89     assert(preparing);
  90     preparing = false;
  91     ch.write(null);
  92     for (t <- workers) t.join();
  93   }
  94
  95   def resolve(name: String, flags: String): Seq[InetAddress] =
  96     map(name).get(flags);
  97 }
  98
  99 def fmtpath(path: Seq[String]) =
 100   path.reverse map { i => s"`$i'" } mkString " -> ";
 101
 102 class ConfigSyntaxError(val file: File, val lno: Int, val msg: String)
 103         extends Exception {
 104   override def getMessage(): String = s"$file:$lno: $msg";
 105 }
 106 class MissingConfigSection(val sect: String) extends Exception {
 107   override def getMessage(): String =
 108     s"missing configuration section `$sect'";
 109 }
 110 class MissingConfigItem(val sect: String, val key: String,
 111                         val path: Seq[(String)]) extends Exception {
 112   override def getMessage(): String = {
 113     val msg = s"missing configuration item `$key' in section `$sect'";
 114     if (path == Nil) msg
 115     else msg + s" (wanted while expanding ${fmtpath(path)})"
 116   }
 117 }
 118 class AmbiguousConfig(val key: String,
 119                       val v0: String, val p0: Seq[String],
 120                       val v1: String, val p1: Seq[String])
 121         extends Exception {
 122   override def getMessage(): String =
 123     s"ambiguous answer resolving key `$key': " +
 124     s"path ${fmtpath(p0)} yields `$v0', but ${fmtpath(p1)} yields `$v1'";
 125 }
 126
 127 class ConfigCycle(val key: String, path: Seq[String]) extends Exception {
 128   override def getMessage(): String =
 129     s"found a cycle ${fmtpath(path)} looking up key `$key'";
 130 }
 131 class NoHostAddresses(val sect: String, val key: String, val host: String)
 132         extends Exception {
 133   override def getMessage(): String =
 134     s"no addresses found for `$host' (key `$key' in section `$sect')";
 135 }
 136
 137 object Config {
 138   sealed abstract class ConfigCacheEntry;
 139   case object StillLooking extends ConfigCacheEntry;
 140   case object NotFound extends ConfigCacheEntry;
 141   case class Found(value: String, path: Seq[String])
 142     extends ConfigCacheEntry;
 143 }
 144
 145 class Config { conf =>
 146   import Config._;
 147   class Section(val name: String) {
 148     val itemmap = HashMap[String, String]();
 149     val cache = HashMap[String, ConfigCacheEntry]();
 150     override def toString: String = s"${getClass.getName}($name)";
 151     def parents: Seq[Section] =
 152       (itemmap.get("@inherit")
 153        map { pp => (RX_PARENT.findAllIn(pp) map { conf.section _ }).toList }
 154        getOrElse Nil);
 155
 156     def get_internal(key: String, path: Seq[String] = Nil):
 157               Option[(String, Seq[String])] = {
 158       val incpath = name +: path;
 159
 160       for (r <- cache.get(key)) r match {
 161         case StillLooking => throw new ConfigCycle(key, incpath)
 162         case NotFound => return None
 163         case Found(v, p) => return Some((v, p ++ path));
 164       }
 165
 166       for (v <- itemmap.get(key)) {
 167         cache(key) = Found(v, Seq(name));
 168         return Some((v, incpath));
 169       }
 170
 171       cache(key) = StillLooking;
 172
 173       ((None: Option[(String, Seq[String])]) /: parents) { (st, parent) =>
 174         parent.get_internal(key, incpath) match {
 175           case None => st;
 176           case newst@Some((v, p)) => st match {
 177             case None => newst
 178             case Some((vv, _)) if v == vv => st
 179             case Some((vv, pp)) =>
 180               throw new AmbiguousConfig(key, v, p, vv, pp)
 181           }
 182         }
 183       } match {
 184         case None => cache(key) = NotFound; None
 185         case Some((v, p)) =>
 186           cache(key) = Found(v, p dropRight path.length);
 187           Some((v, p))
 188       }
 189     }
 190
 191     def get(key: String, resolve: Boolean = true,
 192             path: Seq[String] = Nil): String = {
 193       val v0 = key match {
 194         case "name" => itemmap.getOrElse("name", name)
 195         case _ => get_internal(key).
 196           getOrElse(throw new MissingConfigItem(name, key, path))._1
 197       }
 198       expand(key, v0, resolve, path)
 199     }
 200
 201     def expand(key: String, value: String, resolve: Boolean,
 202                path: Seq[String]): String = {
 203       val v1 = RX_REF.replaceAllIn(value, { m =>
 204         Regex.quoteReplacement(get(m.group(1), resolve, path))
 205       });
 206       val v2 = if (!resolve) v1
 207                else RX_RESOLVE.replaceAllIn(v1, { m =>
 208                  resolver.resolve(m.group(2), m.group(1)) match {
 209                    case Nil =>
 210                      throw new NoHostAddresses(name, key, m.group(2));
 211                    case addrs =>
 212                      Regex.quoteReplacement((addrs map { _.getHostAddress }).
 213                                             mkString(" "));
 214                  }
 215                })
 216       v2
 217     }
 218
 219     def items: Seq[String] = {
 220       val b = Seq.newBuilder[String];
 221       val seen = HashSet[String]();
 222       val visiting = HashSet[String](name);
 223       var stack = List(this);
 224
 225       while (stack != Nil) {
 226         val sect = stack.head; stack = stack.tail;
 227         for (k <- sect.itemmap.keys)
 228           if (!(seen contains k)) { b += k; seen += k; }
 229         for (p <- sect.parents)
 230           if (!(visiting contains p.name))
 231             { stack ::= p; visiting += p.name; }
 232       }
 233       b.result
 234     }
 235   }
 236   val sectmap = new HashMap[String, Section];
 237   def sections: Iterator[Section] = sectmap.values.iterator;
 238   def section(name: String): Section =
 239     sectmap.getOrElse(name, throw new MissingConfigSection(name));
 240
 241   val resolver = new BulkResolver;
 242
 243   def parseFile(path: File): this.type = {
 244 println(s";; parse ${path.getPath}");
 245     withCleaner { clean =>
 246       val in = new FileReader(path); clean { in.close(); }
 247
 248       val lno = 1;
 249       val b = new StringBuilder;
 250       var key: String = null;
 251       var sect: Section = null;
 252       def flush() {
 253         if (key != null) {
 254           sect.itemmap(key) = b.result;
 255 println(s";; in `${sect.name}', set `$key' to `${b.result}'");
 256           b.clear();
 257           key = null;
 258         }
 259       }
 260       for (line <- lines(in)) line match {
 261         case RX_COMMENT() =>
 262           ();
 263         case RX_GRPHDR(grp) =>
 264           flush();
 265           sect = sectmap.getOrElseUpdate(grp, new Section(grp));
 266         case RX_CONT(v) =>
 267           if (key == null) {
 268             throw new ConfigSyntaxError(
 269               path, lno, "no config value to continue");
 270           }
 271           b += '\n'; b ++= v;
 272         case RX_ASSGN(k, v) =>
 273           if (sect == null) {
 274             throw new ConfigSyntaxError(
 275               path, lno, "no active section to update");
 276           }
 277           flush();
 278           key = k; b ++= v;
 279         case _ =>
 280           throw new ConfigSyntaxError(path, lno, "incomprehensible line");
 281       }
 282       flush();
 283     }
 284     this
 285   }
 286   def parse(path: File): this.type = {
 287     if (!path.isDirectory) parseFile(path);
 288     else for {
 289       f <- path.listFiles sortBy { _.getName };
 290       name = f.getName;
 291       if name.length > 0;
 292       tail = name(name.length - 1);
 293       if tail != '#' && tail != '~'
 294     } parseFile(f);
 295     this
 296   }
 297   def parse(path: String): this.type = parse(new File(path));
 298
 299   def analyse() {
 300 println(";; resolving all...");
 301     for ((_, sect) <- sectmap) {
 302 println(s";; resolving in section `${sect.name}'...");
 303       for (key <- sect.items) {
 304 println(s";;    resolving in key `$key'...");
 305         val mm = RX_RESOLVE.findAllIn(sect.get(key, false));
 306         for (host <- mm) { resolver.prepare(mm.group(2)); }
 307       }
 308     }
 309     resolver.finish();
 310
 311     def dumpsect(sect: Section) {
 312       for (k <- sect.items.filterNot(_.startsWith("@")).sorted)
 313         println(s";;    `$k' -> `${sect.get(k)}'")
 314     }
 315     for (sect <- sectmap.values.toSeq sortBy { _.name })
 316       if (sect.name.startsWith("@")) ();
 317       else if (sect.name.startsWith("$")) {
 318         println(s";; special section `${sect.name}'");
 319         dumpsect(sect);
 320       } else {
 321         println(s";; peer section `${sect.name}'");
 322         dumpsect(sect);
 323       }
 324   }
 325 }
 326
 327 }