Simulation of simple clock fusion protocol

clock_sync.kt

import kotlin.math.floor
import kotlin.math.max
import kotlin.math.min
import kotlin.math.round
import kotlin.random.Random

const val N = 10000
const val rho = 0.0
const val minD = 0
const val maxD = 1000
const val ntpPrec = 1000

data class Interval(val a: Double, val b: Double, val mid: Double = (a+b)/2) {
  constructor(a: Int, b: Int): this(a.toDouble(), b.toDouble())
  constructor(a: Double, b: Int): this(a, b.toDouble())
  constructor(a: Int, b: Double): this(a.toDouble(), b)
}

data class Mark(val v: Double, val start: Boolean)

enum class AvgAlgo {
  MARZULLO, IYENGAR_BROOKS, FCA_MID, FCA_MEAN, APPROX_BA
}

fun approx(data: List<Interval>, k: Int): Interval {
  val b = data.map { it.mid }.sorted()
  return Interval(b[k], b[N-k-1])
}

fun fca(data: List<Interval>, k: Int, delta: Double, average: Boolean): Interval {
  val b = data.map { it.mid }.sorted()
  val acceptable = hashSetOf<Double>()
  var m = 0
  for (i in 0 until k) {
    if (b[i+N-k-1] - b[i] <= delta) {
      acceptable.addAll(b.subList(max(m,i), i+N-k))
      m = i+N-k
    }
  }
  val minV = acceptable.min()!!
  val maxV = acceptable.max()!!
  val avg = acceptable.average()
  val mid = (minV + maxV)/2
  val res = if (average) avg else (avg*acceptable.size + (b.size-acceptable.size)*mid)/b.size
  return Interval(minV, maxV, res)
}

fun marzullo(data: List<Interval>, k: Int, iyengarBrooks: Boolean): Interval {
  val b = data
    .flatMap { listOf(
      Mark(it.a-Double.MIN_VALUE, true),
      Mark(it.b+Double.MIN_VALUE, false))}
    .sortedBy { it.v }

  var w = 0
  var minV = Double.MAX_VALUE
  var maxV = -Double.MAX_VALUE
  var sumV = 0.0
  var cntV = 0
  for (i in b.indices) {
    if (i > 0) {
      val start = b[i - 1].v
      val end = b[i].v
      if (start < end && w >= (N-k)) {
        minV = min(minV, start)
        maxV = max(maxV, end)
        sumV += (start+end)*w / 2.0
        cntV += w
      }
    }
    if (b[i].start) {
      w += 1
    } else {
      w -= 1
    }
  }
  if (iyengarBrooks) {
    return Interval(minV, maxV, sumV/cntV)
  } else {
    return Interval(minV, maxV)
  }
}

data class Tick(val time: Int)
data class Msg(val peer: Int, val sendTime: Int, val receiveTime: Int)

data class PeerInfo(val sendTime: Int, val receiveTime: Int)

class Actor(val peer: Int) {
  val peerData = Array<PeerInfo?>(N) { null }
  fun onMsg(msg: Msg) {
    peerData[msg.peer] = PeerInfo(msg.sendTime, msg.receiveTime)
  }
  fun estimate(currTime: Int, avgAlgo: AvgAlgo): Interval {
    val intervals = arrayListOf<Interval>()
    for (i in peerData.indices) {
      val interval = if (i == peer) {
        Interval(currTime - ntpPrec / 2, currTime + ntpPrec / 2)
      } else {
        val d = peerData[i]
        if (d != null) {
          val a = d.sendTime - ntpPrec / 2 + (currTime - d.receiveTime) / (1 + rho) + minD
          val b = d.sendTime + ntpPrec / 2 + (currTime - d.receiveTime) * (1 + rho) + maxD
          Interval(a, b)
        } else {
          Interval(-Double.MAX_VALUE, Double.MAX_VALUE)
        }
      }
      intervals.add(interval)
      //println("$interval ${interval.b-interval.a}")
    }
    val k = floor((N-1)/3.0).toInt()
    return when(avgAlgo) {
      AvgAlgo.MARZULLO -> marzullo(intervals, k, false)
      AvgAlgo.IYENGAR_BROOKS -> marzullo(intervals, k, true)
      AvgAlgo.FCA_MID -> fca(intervals, k, (ntpPrec + maxD-minD).toDouble(), false)
      AvgAlgo.FCA_MEAN -> fca(intervals, k, (ntpPrec + maxD-minD).toDouble(), true)
      AvgAlgo.APPROX_BA -> approx(intervals, k)
    }
  }
  fun onTick(tick: Tick) {

  }
}

fun main(args: Array<String>) {
  val results = AvgAlgo.values().map { it to arrayListOf<Double>() }.toMap()
  for(ii in 0 until 100) {
    for(kk in 5 .. 25) {
      val a = Actor(0)
      val rnd = Random(System.currentTimeMillis())

      val base = 0
      val badGuys = (1 until N).shuffled(rnd).subList(0, (N-1)/3).toSet()
      for (i in 1 until N) {
        val slots = 0 //rnd.nextInt(32)
        val netDelay = rnd.nextInt(minD, maxD)
        val acc = rnd.nextInt(-ntpPrec / 2, ntpPrec / 2 + 1)
        var t = base - slots * 12 + acc + netDelay
        if (i in badGuys) {
          t -= (ntpPrec * kk * 0.1).toInt()
          //t += ntpPrec * rnd.nextInt(1,2)
        }
        a.onMsg(Msg(i, base - slots * 12, t))
      }
      println("${kk*0.1} ----")
      for(algo in AvgAlgo.values()) {
        val res = a.estimate(base, avgAlgo = algo)
        println("$algo $res")

        (results[algo] ?: error("")).add(res.mid)
      }
    }
  }
  println("----------------")
  for(k in results.keys) {
    println("$k ${round(results[k]!!.max()!!)}")
  }
}