text:
package org.apache.s2graph.core.storage

import org.apache.s2graph.core.GraphExceptions.{NoStackException, FetchTimeoutException}
import org.apache.s2graph.core._
import org.apache.s2graph.core.mysqls.{Label, LabelMeta}
import org.apache.s2graph.core.parsers.WhereParser
import org.apache.s2graph.core.storage.serde.indexedge.wide.{IndexEdgeDeserializable, IndexEdgeSerializable}
import org.apache.s2graph.core.storage.serde.snapshotedge.wide.SnapshotEdgeDeserializable
import org.apache.s2graph.core.storage.serde.vertex.{VertexDeserializable, VertexSerializable}
import org.apache.s2graph.core.types._
import org.apache.s2graph.core.utils.{DeferCache, Extensions, logger}

import scala.collection.mutable.{ArrayBuffer, ListBuffer}
import scala.concurrent.{ExecutionContext, Future, Promise}
import scala.util.{Random, Try}
import java.util.concurrent.{Executors, TimeUnit}

import com.typesafe.config.{Config, ConfigFactory}
import org.apache.hadoop.hbase.util.Bytes

abstract class Storage[Q, R](val graph: S2Graph,
                          val config: Config)(implicit ec: ExecutionContext) {
  import HBaseType._
  import S2Graph._

  val BackoffTimeout = graph.BackoffTimeout
  val MaxRetryNum = graph.MaxRetryNum
  val MaxBackOff = graph.MaxBackOff
  val FailProb = graph.FailProb
  val LockExpireDuration =  graph.LockExpireDuration
  val MaxSize = graph.MaxSize
  val ExpireAfterWrite = graph.ExpireAfterWrite
  val ExpireAfterAccess = graph.ExpireAfterAccess

  /** retry scheduler */
  val scheduledThreadPool = Executors.newSingleThreadScheduledExecutor()

  /**
   * Compatibility table
   * | label schema version | snapshot edge | index edge | vertex | note |
   * | v1 | serde.snapshotedge.wide | serde.indexedge.wide | serde.vertex | do not use this. this exist only for backward compatibility issue |
   * | v2 | serde.snapshotedge.wide | serde.indexedge.wide | serde.vertex | do not use this. this exist only for backward compatibility issue |
   * | v3 | serde.snapshotedge.tall | serde.indexedge.wide | serde.vertex | recommended with HBase. current stable schema |
   * | v4 | serde.snapshotedge.tall | serde.indexedge.tall | serde.vertex | experimental schema. use scanner instead of get |
   *
   */

  /**
   * create serializer that knows how to convert given snapshotEdge into kvs: Seq[SKeyValue]
   * so we can store this kvs.
   * @param snapshotEdge: snapshotEdge to serialize
   * @return serializer implementation for StorageSerializable which has toKeyValues return Seq[SKeyValue]
   */
  def snapshotEdgeSerializer(snapshotEdge: SnapshotEdge): Serializable[SnapshotEdge] = {
    snapshotEdge.schemaVer match {
//      case VERSION1 |
      case VERSION2 => new serde.snapshotedge.wide.SnapshotEdgeSerializable(snapshotEdge)
      case VERSION3 | VERSION4 => new serde.snapshotedge.tall.SnapshotEdgeSerializable(snapshotEdge)
      case _ => throw new RuntimeException(s"not supported version: ${snapshotEdge.schemaVer}")
    }
  }

  /**
   * create serializer that knows how to convert given indexEdge into kvs: Seq[SKeyValue]
   * @param indexEdge: indexEdge to serialize
   * @return serializer implementation
   */
  def indexEdgeSerializer(indexEdge: IndexEdge): Serializable[IndexEdge] = {
    indexEdge.schemaVer match {
//      case VERSION1
      case VERSION2 | VERSION3 => new IndexEdgeSerializable(indexEdge)
      case VERSION4 => new serde.indexedge.tall.IndexEdgeSerializable(indexEdge)
      case _ => throw new RuntimeException(s"not supported version: ${indexEdge.schemaVer}")

    }
  }

  /**
   * create serializer that knows how to convert given vertex into kvs: Seq[SKeyValue]
   * @param vertex: vertex to serialize
   * @return serializer implementation
   */
  def vertexSerializer(vertex: S2Vertex): Serializable[S2Vertex] = new VertexSerializable(vertex)

  /**
   * create deserializer that can parse stored CanSKeyValue into snapshotEdge.
   * note that each storage implementation should implement implicit type class
   * to convert storage dependent dataType into common SKeyValue type by implementing CanSKeyValue
   *
   * ex) Asynchbase use it's KeyValue class and CanSKeyValue object has implicit type conversion method.
   * if any storaage use different class to represent stored byte array,
   * then that storage implementation is responsible to provide implicit type conversion method on CanSKeyValue.
   * */

  val snapshotEdgeDeserializer: Deserializable[SnapshotEdge] = new serde.snapshotedge.tall.SnapshotEdgeDeserializable(graph)

  def snapshotEdgeDeserializer(schemaVer: String): Deserializable[SnapshotEdge] = snapshotEdgeDeserializer

  /** create deserializer that can parse stored CanSKeyValue into indexEdge. */
  val indexEdgeDeserializer: Deserializable[S2Edge] = new serde.indexedge.tall.IndexEdgeDeserializable(graph)

  def indexEdgeDeserializer(schemaVer: String) = new serde.indexedge.tall.IndexEdgeDeserializable(graph)

  /** create deserializer that can parser stored CanSKeyValue into vertex. */
  val vertexDeserializer: Deserializable[S2Vertex] = new VertexDeserializable(graph)

  /**
   * decide how to store given key values Seq[SKeyValue] into storage using storage's client.
   * note that this should be return true on all success.
   * we assumes that each storage implementation has client as member variable.
   *
   *
   * @param cluster: where this key values should be stored.
   * @param kvs: sequence of SKeyValue that need to be stored in storage.
   * @param withWait: flag to control wait ack from storage.
   *                  note that in AsynchbaseStorage(which support asynchronous operations), even with true,
   *                  it never block thread, but rather submit work and notified by event loop when storage send ack back.
   * @return ack message from storage.
   */
  def writeToStorage(cluster: String, kvs: Seq[SKeyValue], withWait: Boolean): Future[Boolean]

//  def writeToStorage(kv: SKeyValue, withWait: Boolean): Future[Boolean]

  /**
   * fetch SnapshotEdge for given request from storage.
   * also storage datatype should be converted into SKeyValue.
   * note that return type is Sequence rather than single SKeyValue for simplicity,
   * even though there is assertions sequence.length == 1.
   * @param request
   * @return
   */
  def fetchSnapshotEdgeKeyValues(request: QueryRequest): Future[Seq[SKeyValue]]

  /**
   * write requestKeyValue into storage if the current value in storage that is stored matches.
   * note that we only use SnapshotEdge as place for lock, so this method only change SnapshotEdge.
   *
   * Most important thing is this have to be 'atomic' operation.
   * When this operation is mutating requestKeyValue's snapshotEdge, then other thread need to be
   * either blocked or failed on write-write conflict case.
   *
   * Also while this method is still running, then fetchSnapshotEdgeKeyValues should be synchronized to
   * prevent wrong data for read.
   *
   * Best is use storage's concurrency control(either pessimistic or optimistic) such as transaction,
   * compareAndSet to synchronize.
   *
   * for example, AsynchbaseStorage use HBase's CheckAndSet atomic operation to guarantee 'atomicity'.
   * for storage that does not support concurrency control, then storage implementation
   * itself can maintain manual locks that synchronize read(fetchSnapshotEdgeKeyValues)
   * and write(writeLock).
   * @param requestKeyValue
   * @param expectedOpt
   * @return
   */
  def writeLock(requestKeyValue: SKeyValue, expectedOpt: Option[SKeyValue]): Future[Boolean]

  /**
   * build proper request which is specific into storage to call fetchIndexEdgeKeyValues or fetchSnapshotEdgeKeyValues.
   * for example, Asynchbase use GetRequest, Scanner so this method is responsible to build
   * client request(GetRequest, Scanner) based on user provided query.
    *
    * @param queryRequest
   * @return
    */
  protected def buildRequest(queryRequest: QueryRequest, edge: S2Edge): Q

  /**
   * fetch IndexEdges for given queryParam in queryRequest.
   * this expect previous step starting score to propagate score into next step.
   * also parentEdges is necessary to return full bfs tree when query require it.
   *
   * note that return type is general type.
   * for example, currently we wanted to use Asynchbase
   * so single I/O return type should be Deferred[T].
   *
   * if we use native hbase client, then this return type can be Future[T] or just T.
    *
    * @param queryRequest
   * @param isInnerCall
   * @param parentEdges
   * @return
   */
  def fetch(queryRequest: QueryRequest,
            isInnerCall: Boolean,
            parentEdges: Seq[EdgeWithScore]): R

  /**
   * responsible to fire parallel fetch call into storage and create future that will return merged result.
   *
   * @param queryRequests
   * @param prevStepEdges
   * @return
   */
  def fetches(queryRequests: Seq[QueryRequest],
              prevStepEdges: Map[VertexId, Seq[EdgeWithScore]]): Future[Seq[StepResult]]

  /**
   * fetch Vertex for given request from storage.
    *
    * @param request
   * @return
   */
  def fetchVertexKeyValues(request: QueryRequest): Future[Seq[SKeyValue]]

  /**
   * decide how to apply given edges(indexProps values + Map(_count -> countVal)) into storage.
    *
    * @param edges
   * @param withWait
   * @return
   */
  def incrementCounts(edges: Seq[S2Edge], withWait: Boolean): Future[Seq[(Boolean, Long, Long)]]

  /**
   * this method need to be called when client shutdown. this is responsible to cleanUp the resources
   * such as client into storage.
   */
  def flush(): Unit = {
  }

  def fetchEdgesAll(): Future[Seq[S2Edge]]

  def fetchVerticesAll(): Future[Seq[S2Vertex]]

  /**
   * create table on storage.
   * if storage implementation does not support namespace or table, then there is nothing to be done
    *
    * @param zkAddr
   * @param tableName
   * @param cfs
   * @param regionMultiplier
   * @param ttl
   * @param compressionAlgorithm
   */
  def createTable(zkAddr: String,
                  tableName: String,
                  cfs: List[String],
                  regionMultiplier: Int,
                  ttl: Option[Int],
                  compressionAlgorithm: String,
                  replicationScopeOpt: Option[Int] = None,
                  totalRegionCount: Option[Int] = None): Unit

  def truncateTable(zkAddr: String, tableNameStr: String): Unit = {}

  def deleteTable(zkAddr: String, tableNameStr: String): Unit = {}

  def shutdown(): Unit

  /** Public Interface */
  def getVertices(vertices: Seq[S2Vertex]): Future[Seq[S2Vertex]] = {
    def fromResult(kvs: Seq[SKeyValue],
                   version: String): Option[S2Vertex] = {
      if (kvs.isEmpty) None
      else vertexDeserializer.fromKeyValues(kvs, None)
//        .map(S2Vertex(graph, _))
    }

    val futures = vertices.map { vertex =>
      val queryParam = QueryParam.Empty
      val q = Query.toQuery(Seq(vertex), Seq(queryParam))
      val queryRequest = QueryRequest(q, stepIdx = -1, vertex, queryParam)

      fetchVertexKeyValues(queryRequest).map { kvs =>
        fromResult(kvs, vertex.serviceColumn.schemaVersion)
      } recoverWith { case ex: Throwable =>
        Future.successful(None)
      }
    }

    Future.sequence(futures).map { result => result.toList.flatten }
  }
  def mutateStrongEdges(_edges: Seq[S2Edge], withWait: Boolean): Future[Seq[Boolean]] = {

    val edgeWithIdxs = _edges.zipWithIndex
    val grouped = edgeWithIdxs.groupBy { case (edge, idx) =>
      (edge.innerLabel, edge.srcVertex.innerId, edge.tgtVertex.innerId)
    } toSeq

    val mutateEdges = grouped.map { case ((_, _, _), edgeGroup) =>
      val edges = edgeGroup.map(_._1)
      val idxs = edgeGroup.map(_._2)
      // After deleteAll, process others
      val mutateEdgeFutures = edges.toList match {
        case head :: tail =>
          val edgeFuture = mutateEdgesInner(edges, checkConsistency = true , withWait)

          //TODO: decide what we will do on failure on vertex put
          val puts = buildVertexPutsAsync(head)
          val vertexFuture = writeToStorage(head.innerLabel.hbaseZkAddr, puts, withWait)
          Seq(edgeFuture, vertexFuture)
        case Nil => Nil
      }

      val composed = for {
//        deleteRet <- Future.sequence(deleteAllFutures)
        mutateRet <- Future.sequence(mutateEdgeFutures)
      } yield mutateRet

      composed.map(_.forall(identity)).map { ret => idxs.map( idx => idx -> ret) }
    }

    Future.sequence(mutateEdges).map { squashedRets =>
      squashedRets.flatten.sortBy { case (idx, ret) => idx }.map(_._2)
    }
  }

  def mutateVertex(vertex: S2Vertex, withWait: Boolean): Future[Boolean] = {
    if (vertex.op == GraphUtil.operations("delete")) {
      writeToStorage(vertex.hbaseZkAddr,
        vertexSerializer(vertex).toKeyValues.map(_.copy(operation = SKeyValue.Delete)), withWait)
    } else if (vertex.op == GraphUtil.operations("deleteAll")) {
      logger.info(s"deleteAll for vertex is truncated. $vertex")
      Future.successful(true) // Ignore withWait parameter, because deleteAll operation may takes long time
    } else {
      writeToStorage(vertex.hbaseZkAddr, buildPutsAll(vertex), withWait)
    }
  }

  def mutateVertices(vertices: Seq[S2Vertex],
                     withWait: Boolean = false): Future[Seq[Boolean]] = {
    val futures = vertices.map { vertex => mutateVertex(vertex, withWait) }
    Future.sequence(futures)
  }

  def mutateEdgesInner(edges: Seq[S2Edge],
                       checkConsistency: Boolean,
                       withWait: Boolean): Future[Boolean] = {
    assert(edges.nonEmpty)
    // TODO:: remove after code review: unreachable code
    if (!checkConsistency) {

      val zkQuorum = edges.head.innerLabel.hbaseZkAddr
      val futures = edges.map { edge =>
        val (_, edgeUpdate) = S2Edge.buildOperation(None, Seq(edge))

        val (bufferIncr, nonBufferIncr) = increments(edgeUpdate.deepCopy)
        val mutations =
          indexedEdgeMutations(edgeUpdate.deepCopy) ++ snapshotEdgeMutations(edgeUpdate.deepCopy) ++ nonBufferIncr

        if (bufferIncr.nonEmpty) writeToStorage(zkQuorum, bufferIncr, withWait = false)

        writeToStorage(zkQuorum, mutations, withWait)
      }
      Future.sequence(futures).map { rets => rets.forall(identity) }
    } else {
      fetchSnapshotEdgeInner(edges.head).flatMap { case (queryParam, snapshotEdgeOpt, kvOpt) =>
        retry(1)(edges, 0, snapshotEdgeOpt)
      }
    }
  }

  def exponentialBackOff(tryNum: Int) = {
    // time slot is divided by 10 ms
    val slot = 10
    Random.nextInt(Math.min(BackoffTimeout, slot * Math.pow(2, tryNum)).toInt)
  }

  def retry(tryNum: Int)(edges: Seq[S2Edge], statusCode: Byte, fetchedSnapshotEdgeOpt: Option[S2Edge]): Future[Boolean] = {
    if (tryNum >= MaxRetryNum) {
      edges.foreach { edge =>
        logger.error(s"commit failed after $MaxRetryNum\n${edge.toLogString}")
      }

      Future.successful(false)
    } else {
      val future = commitUpdate(edges, statusCode, fetchedSnapshotEdgeOpt)
      future.onSuccess {
        case success =>
          logger.debug(s"Finished. [$tryNum]\n${edges.head.toLogString}\n")
      }
      future recoverWith {
        case FetchTimeoutException(retryEdge) =>
          logger.info(s"[Try: $tryNum], Fetch fail.\n${retryEdge}")
          /* fetch failed. re-fetch should be done */
          fetchSnapshotEdgeInner(edges.head).flatMap { case (queryParam, snapshotEdgeOpt, kvOpt) =>
            retry(tryNum + 1)(edges, statusCode, snapshotEdgeOpt)
          }

        case PartialFailureException(retryEdge, failedStatusCode, faileReason) =>
          val status = failedStatusCode match {
            case 0 => "AcquireLock failed."
            case 1 => "Mutation failed."
            case 2 => "Increment failed."
            case 3 => "ReleaseLock failed."
            case 4 => "Unknown"
          }
          logger.info(s"[Try: $tryNum], [Status: $status] partial fail.\n${retryEdge.toLogString}\nFailReason: ${faileReason}")

          /* retry logic */
          val promise = Promise[Boolean]
          val backOff = exponentialBackOff(tryNum)
          scheduledThreadPool.schedule(new Runnable {
            override def run(): Unit = {
              val future = if (failedStatusCode == 0) {
                // acquire Lock failed. other is mutating so this thead need to re-fetch snapshotEdge.
                /* fetch failed. re-fetch should be done */
                fetchSnapshotEdgeInner(edges.head).flatMap { case (queryParam, snapshotEdgeOpt, kvOpt) =>
                  retry(tryNum + 1)(edges, statusCode, snapshotEdgeOpt)
                }
              } else {
                // partial failure occur while self locked and mutating.
                //            assert(fetchedSnapshotEdgeOpt.nonEmpty)
                retry(tryNum + 1)(edges, failedStatusCode, fetchedSnapshotEdgeOpt)
              }
              promise.completeWith(future)
            }

          }, backOff, TimeUnit.MILLISECONDS)
          promise.future

        case ex: Exception =>
          logger.error("Unknown exception", ex)
          Future.successful(false)
      }
    }
  }

  protected def commitUpdate(edges: Seq[S2Edge],
                             statusCode: Byte,
                             fetchedSnapshotEdgeOpt: Option[S2Edge]): Future[Boolean] = {
//    Future.failed(new PartialFailureException(edges.head, 0, "ahahah"))
    assert(edges.nonEmpty)
//    assert(statusCode == 0 || fetchedSnapshotEdgeOpt.isDefined)

    statusCode match {
      case 0 =>
        fetchedSnapshotEdgeOpt match {
          case None =>
          /*
           * no one has never mutated this SN.
           * (squashedEdge, edgeMutate) = Edge.buildOperation(None, edges)
           * pendingE = squashedEdge.copy(statusCode = 1, lockTs = now, version = squashedEdge.ts + 1)
           * lock = (squashedEdge, pendingE)
           * releaseLock = (edgeMutate.newSnapshotEdge, None)
           */
            val (squashedEdge, edgeMutate) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, edges)

            assert(edgeMutate.newSnapshotEdge.isDefined)

            val lockTs = Option(System.currentTimeMillis())
            val pendingEdge = squashedEdge.copy(statusCode = 1, lockTs = lockTs, version = squashedEdge.ts + 1)
            val lockSnapshotEdge = squashedEdge.toSnapshotEdge.copy(pendingEdgeOpt = Option(pendingEdge))
            val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge.get.copy(statusCode = 0,
              pendingEdgeOpt = None, version = lockSnapshotEdge.version + 1)

            commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)

          case Some(snapshotEdge) =>
            snapshotEdge.pendingEdgeOpt match {
              case None =>
                /*
                 * others finished commit on this SN. but there is no contention.
                 * (squashedEdge, edgeMutate) = Edge.buildOperation(snapshotEdgeOpt, edges)
                 * pendingE = squashedEdge.copy(statusCode = 1, lockTs = now, version = snapshotEdge.version + 1) ?
                 * lock = (snapshotEdge, pendingE)
                 * releaseLock = (edgeMutate.newSnapshotEdge, None)
                 */
                val (squashedEdge, edgeMutate) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, edges)
                if (edgeMutate.newSnapshotEdge.isEmpty) {
                  logger.debug(s"drop this requests: \n${edges.map(_.toLogString).mkString("\n")}")
                  Future.successful(true)
                } else {
                  val lockTs = Option(System.currentTimeMillis())
                  val pendingEdge = squashedEdge.copy(statusCode = 1, lockTs = lockTs, version = snapshotEdge.version + 1)
                  val lockSnapshotEdge = snapshotEdge.toSnapshotEdge.copy(pendingEdgeOpt = Option(pendingEdge))
                  val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge.get.copy(statusCode = 0,
                    pendingEdgeOpt = None, version = lockSnapshotEdge.version + 1)
                  commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)
                }
              case Some(pendingEdge) =>
                val isLockExpired = pendingEdge.lockTs.get + LockExpireDuration < System.currentTimeMillis()
                if (isLockExpired) {
                  /*
                   * if pendingEdge.ts == snapshotEdge.ts =>
                   *    (squashedEdge, edgeMutate) = Edge.buildOperation(None, Seq(pendingEdge))
                   * else =>
                   *    (squashedEdge, edgeMutate) = Edge.buildOperation(snapshotEdgeOpt, Seq(pendingEdge))
                   * pendingE = squashedEdge.copy(statusCode = 1, lockTs = now, version = snapshotEdge.version + 1)
                   * lock = (snapshotEdge, pendingE)
                   * releaseLock = (edgeMutate.newSnapshotEdge, None)
                   */
                  logger.debug(s"${pendingEdge.toLogString} has been expired.")
                  val (squashedEdge, edgeMutate) =
                    if (pendingEdge.ts == snapshotEdge.ts) S2Edge.buildOperation(None, pendingEdge +: edges)
                    else S2Edge.buildOperation(fetchedSnapshotEdgeOpt, pendingEdge +: edges)

                  val lockTs = Option(System.currentTimeMillis())
                  val newPendingEdge = squashedEdge.copy(statusCode = 1, lockTs = lockTs, version = snapshotEdge.version + 1)
                  val lockSnapshotEdge = snapshotEdge.toSnapshotEdge.copy(pendingEdgeOpt = Option(newPendingEdge))
                  val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge.get.copy(statusCode = 0,
                    pendingEdgeOpt = None, version = lockSnapshotEdge.version + 1)

                  commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)
                } else {
                  /*
                   * others finished commit on this SN and there is currently contention.
                   * this can't be proceed so retry from re-fetch.
                   * throw EX
                   */
                  val (squashedEdge, _) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, edges)
                  Future.failed(new PartialFailureException(squashedEdge, 0, s"others[${pendingEdge.ts}] is mutating. me[${squashedEdge.ts}]"))
                }
            }

        }
      case _ =>

        /*
         * statusCode > 0 which means self locked and there has been partial failure either on mutate, increment, releaseLock
         */

        /*
         * this succeed to lock this SN. keep doing on commit process.
         * if SN.isEmpty =>
         * no one never succed to commit on this SN.
         * this is first mutation try on this SN.
         * (squashedEdge, edgeMutate) = Edge.buildOperation(SN, edges)
         * else =>
         * assert(SN.pengingEdgeOpt.isEmpty) no-fetch after acquire lock when self retrying.
         * there has been success commit on this SN.
         * (squashedEdge, edgeMutate) = Edge.buildOperation(SN, edges)
         * releaseLock = (edgeMutate.newSnapshotEdge, None)
         */
        val _edges =
          if (fetchedSnapshotEdgeOpt.isDefined && fetchedSnapshotEdgeOpt.get.pendingEdgeOpt.isDefined) fetchedSnapshotEdgeOpt.get.pendingEdgeOpt.get +: edges
          else edges
        val (squashedEdge, edgeMutate) = S2Edge.buildOperation(fetchedSnapshotEdgeOpt, _edges)
        val newVersion = fetchedSnapshotEdgeOpt.map(_.version).getOrElse(squashedEdge.ts) + 2
        val releaseLockSnapshotEdge = edgeMutate.newSnapshotEdge match {
          case None => squashedEdge.toSnapshotEdge.copy(statusCode = 0, pendingEdgeOpt = None, version = newVersion)
          case Some(newSnapshotEdge) => newSnapshotEdge.copy(statusCode = 0, pendingEdgeOpt = None, version = newVersion)
        }
        // lockSnapshotEdge will be ignored.
        commitProcess(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, releaseLockSnapshotEdge, releaseLockSnapshotEdge, edgeMutate)
    }
  }
  /**
   * orchestrate commit process.
   * we separate into 4 step to avoid duplicating each step over and over.
    *
    * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge.
   * @param fetchedSnapshotEdgeOpt: fetched snapshotEdge from storage before commit process begin.
   * @param lockSnapshotEdge: lockEdge that hold necessary data to lock this snapshotEdge for this thread.
   * @param releaseLockSnapshotEdge: releaseLockEdge that will remove lock by storing new final merged states
   *                               all from current request edges and fetched snapshotEdge.
   * @param edgeMutate: mutations for indexEdge and snapshotEdge.
   * @return
   */
  protected def commitProcess(statusCode: Byte,
                              squashedEdge: S2Edge,
                              fetchedSnapshotEdgeOpt:Option[S2Edge],
                              lockSnapshotEdge: SnapshotEdge,
                              releaseLockSnapshotEdge: SnapshotEdge,
                              edgeMutate: EdgeMutate): Future[Boolean] = {
    for {
      locked <- acquireLock(statusCode, squashedEdge, fetchedSnapshotEdgeOpt, lockSnapshotEdge)
      mutated <- commitIndexEdgeMutations(locked, statusCode, squashedEdge, edgeMutate)
      incremented <- commitIndexEdgeDegreeMutations(mutated, statusCode, squashedEdge, edgeMutate)
      lockReleased <- releaseLock(incremented, statusCode, squashedEdge, releaseLockSnapshotEdge)
    } yield lockReleased
  }

  case class PartialFailureException(edge: S2Edge, statusCode: Byte, failReason: String) extends NoStackException(failReason)

  protected def debug(ret: Boolean, phase: String, snapshotEdge: SnapshotEdge) = {
    val msg = Seq(s"[$ret] [$phase]", s"${snapshotEdge.toLogString()}").mkString("\n")
    logger.debug(msg)
  }

  protected def debug(ret: Boolean, phase: String, snapshotEdge: SnapshotEdge, edgeMutate: EdgeMutate) = {
    val msg = Seq(s"[$ret] [$phase]", s"${snapshotEdge.toLogString()}",
      s"${edgeMutate.toLogString}").mkString("\n")
    logger.debug(msg)
  }

  /**
   * try to acquire lock on storage for this given snapshotEdge(lockEdge).
    *
    * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param fetchedSnapshotEdgeOpt: fetched snapshot edge from storage.
   * @param lockEdge: lockEdge to build RPC request(compareAndSet) into Storage.
   * @return
   */
  protected def acquireLock(statusCode: Byte,
                            squashedEdge: S2Edge,
                            fetchedSnapshotEdgeOpt: Option[S2Edge],
                            lockEdge: SnapshotEdge): Future[Boolean] = {
    if (statusCode >= 1) {
      logger.debug(s"skip acquireLock: [$statusCode]\n${squashedEdge.toLogString}")
      Future.successful(true)
    } else {
      val p = Random.nextDouble()
      if (p < FailProb) {
        Future.failed(new PartialFailureException(squashedEdge, 0, s"$p"))
      } else {
        val lockEdgePut = snapshotEdgeSerializer(lockEdge).toKeyValues.head
        val oldPut = fetchedSnapshotEdgeOpt.map(e => snapshotEdgeSerializer(e.toSnapshotEdge).toKeyValues.head)
        writeLock(lockEdgePut, oldPut).recoverWith { case ex: Exception =>
          logger.error(s"AcquireLock RPC Failed.")
          throw new PartialFailureException(squashedEdge, 0, "AcquireLock RPC Failed")
        }.map { ret =>
          if (ret) {
            val log = Seq(
              "\n",
              "=" * 50,
              s"[Success]: acquireLock",
              s"[RequestEdge]: ${squashedEdge.toLogString}",
              s"[LockEdge]: ${lockEdge.toLogString()}",
              s"[PendingEdge]: ${lockEdge.pendingEdgeOpt.map(_.toLogString).getOrElse("")}",
              "=" * 50, "\n").mkString("\n")

            logger.debug(log)
            //            debug(ret, "acquireLock", edge.toSnapshotEdge)
          } else {
            throw new PartialFailureException(squashedEdge, 0, "hbase fail.")
          }
          true
        }
      }
    }
  }

  /**
   * change this snapshot's state on storage from locked into committed by
   * storing new merged states on storage. merge state come from releaseLockEdge.
   * note that releaseLock return Future.failed on predicate failure.
    *
    * @param predicate: indicate if this releaseLock phase should be proceed or not.
   * @param statusCode: releaseLock do not use statusCode, only for debug.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param releaseLockEdge: final merged states if all process goes well.
   * @return
   */
  protected def releaseLock(predicate: Boolean,
                            statusCode: Byte,
                            squashedEdge: S2Edge,
                            releaseLockEdge: SnapshotEdge): Future[Boolean] = {
    if (!predicate) {
      Future.failed(new PartialFailureException(squashedEdge, 3, "predicate failed."))
    } else {
      val p = Random.nextDouble()
      if (p < FailProb) Future.failed(new PartialFailureException(squashedEdge, 3, s"$p"))
      else {
        val releaseLockEdgePuts = snapshotEdgeSerializer(releaseLockEdge).toKeyValues
        writeToStorage(squashedEdge.innerLabel.hbaseZkAddr, releaseLockEdgePuts, withWait = true).recoverWith {
          case ex: Exception =>
            logger.error(s"ReleaseLock RPC Failed.")
            throw new PartialFailureException(squashedEdge, 3, "ReleaseLock RPC Failed")
        }.map { ret =>
          if (ret) {
            debug(ret, "releaseLock", squashedEdge.toSnapshotEdge)
          } else {
            val msg = Seq("\nFATAL ERROR\n",
              "=" * 50,
              squashedEdge.toLogString,
              releaseLockEdgePuts,
              "=" * 50,
              "\n"
            )
            logger.error(msg.mkString("\n"))
            //          error(ret, "releaseLock", edge.toSnapshotEdge)
            throw new PartialFailureException(squashedEdge, 3, "hbase fail.")
          }
          true
        }
      }
    }
  }

  /**
   *
   * @param predicate: indicate if this commitIndexEdgeMutations phase should be proceed or not.
   * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param edgeMutate: actual collection of mutations. note that edgeMutate contains snapshotEdge mutations,
   *                  but in here, we only use indexEdge's mutations.
   * @return
   */
  protected def commitIndexEdgeMutations(predicate: Boolean,
                       statusCode: Byte,
                       squashedEdge: S2Edge,
                       edgeMutate: EdgeMutate): Future[Boolean] = {
    if (!predicate) Future.failed(new PartialFailureException(squashedEdge, 1, "predicate failed."))
    else {
      if (statusCode >= 2) {
        logger.debug(s"skip mutate: [$statusCode]\n${squashedEdge.toLogString}")
        Future.successful(true)
      } else {
        val p = Random.nextDouble()
        if (p < FailProb) Future.failed(new PartialFailureException(squashedEdge, 1, s"$p"))
        else
          writeToStorage(squashedEdge.innerLabel.hbaseZkAddr, indexedEdgeMutations(edgeMutate), withWait = true).map { ret =>
            if (ret) {
              debug(ret, "mutate", squashedEdge.toSnapshotEdge, edgeMutate)
            } else {
              throw new PartialFailureException(squashedEdge, 1, "hbase fail.")
            }
            true
          }
      }
    }
  }

  /**
   *
   * @param predicate: indicate if this commitIndexEdgeMutations phase should be proceed or not.
   * @param statusCode: current statusCode of this thread to process edges.
   * @param squashedEdge: squashed(in memory) final edge from input edges on same snapshotEdge. only for debug
   * @param edgeMutate: actual collection of mutations. note that edgeMutate contains snapshotEdge mutations,
   *                  but in here, we only use indexEdge's degree mutations.
   * @return
   */
  protected def commitIndexEdgeDegreeMutations(predicate: Boolean,
                          statusCode: Byte,
                          squashedEdge: S2Edge,
                          edgeMutate: EdgeMutate): Future[Boolean] = {

    def _write(kvs: Seq[SKeyValue], withWait: Boolean): Future[Boolean] = {
      writeToStorage(squashedEdge.innerLabel.hbaseZkAddr, kvs, withWait = withWait).map { ret =>
        if (ret) {
          debug(ret, "increment", squashedEdge.toSnapshotEdge, edgeMutate)
        } else {
          throw new PartialFailureException(squashedEdge, 2, "hbase fail.")
        }
        true
      }
    }

    if (!predicate) Future.failed(new PartialFailureException(squashedEdge, 2, "predicate failed."))
    if (statusCode >= 3) {
      logger.debug(s"skip increment: [$statusCode]\n${squashedEdge.toLogString}")
      Future.successful(true)
    } else {
      val p = Random.nextDouble()
      if (p < FailProb) Future.failed(new PartialFailureException(squashedEdge, 2, s"$p"))
      else {
        val (bufferIncr, nonBufferIncr) = increments(edgeMutate.deepCopy)

        if (bufferIncr.nonEmpty) _write(bufferIncr, withWait = false)
        _write(nonBufferIncr, withWait = true)
      }
    }
  }

  /** end of methods for consistency */

  def mutateLog(snapshotEdgeOpt: Option[S2Edge], edges: Seq[S2Edge],
                newEdge: S2Edge, edgeMutate: EdgeMutate) =
    Seq("----------------------------------------------",
      s"SnapshotEdge: ${snapshotEdgeOpt.map(_.toLogString)}",
      s"requestEdges: ${edges.map(_.toLogString).mkString("\n")}",
      s"newEdge: ${newEdge.toLogString}",
      s"mutation: \n${edgeMutate.toLogString}",
      "----------------------------------------------").mkString("\n")

  /** Delete All */
  def deleteAllFetchedEdgesAsyncOld(stepInnerResult: StepResult,
                                              requestTs: Long,
                                              retryNum: Int): Future[Boolean] = {
    if (stepInnerResult.isEmpty) Future.successful(true)
    else {
      val head = stepInnerResult.edgeWithScores.head
      val zkQuorum = head.edge.innerLabel.hbaseZkAddr
      val futures = for {
        edgeWithScore <- stepInnerResult.edgeWithScores
      } yield {
          val edge = edgeWithScore.edge
          val score = edgeWithScore.score

          val edgeSnapshot = edge.copyEdge(propsWithTs = S2Edge.propsToState(edge.updatePropsWithTs()))
          val reversedSnapshotEdgeMutations = snapshotEdgeSerializer(edgeSnapshot.toSnapshotEdge).toKeyValues.map(_.copy(operation = SKeyValue.Put))

          val edgeForward = edge.copyEdge(propsWithTs = S2Edge.propsToState(edge.updatePropsWithTs()))
          val forwardIndexedEdgeMutations = edgeForward.edgesWithIndex.flatMap { indexEdge =>
            indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Delete)) ++
              buildIncrementsAsync(indexEdge, -1L)
          }

          /* reverted direction */
          val edgeRevert = edge.copyEdge(propsWithTs = S2Edge.propsToState(edge.updatePropsWithTs()))
          val reversedIndexedEdgesMutations = edgeRevert.duplicateEdge.edgesWithIndex.flatMap { indexEdge =>
            indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Delete)) ++
              buildIncrementsAsync(indexEdge, -1L)
          }

          val mutations = reversedIndexedEdgesMutations ++ reversedSnapshotEdgeMutations ++ forwardIndexedEdgeMutations

          writeToStorage(zkQuorum, mutations, withWait = true)
        }

      Future.sequence(futures).map { rets => rets.forall(identity) }
    }
  }

  /** End Of Delete All */

  /** Parsing Logic: parse from kv from Storage into Edge */
  def toEdge[K: CanSKeyValue](kv: K,
                              queryRequest: QueryRequest,
                              cacheElementOpt: Option[S2Edge],
                              parentEdges: Seq[EdgeWithScore]): Option[S2Edge] = {
    logger.debug(s"toEdge: $kv")

    try {
      val queryOption = queryRequest.query.queryOption
      val queryParam = queryRequest.queryParam
      val schemaVer = queryParam.label.schemaVersion
      val indexEdgeOpt = indexEdgeDeserializer(schemaVer).fromKeyValues(Seq(kv), cacheElementOpt)
      if (!queryOption.returnTree) indexEdgeOpt.map(indexEdge => indexEdge.copy(parentEdges = parentEdges))
      else indexEdgeOpt
    } catch {
      case ex: Exception =>
        logger.error(s"Fail on toEdge: ${kv.toString}, ${queryRequest}", ex)
        None
    }
  }

  def toSnapshotEdge[K: CanSKeyValue](kv: K,
                                      queryRequest: QueryRequest,
                                      cacheElementOpt: Option[SnapshotEdge] = None,
                                      isInnerCall: Boolean,
                                      parentEdges: Seq[EdgeWithScore]): Option[S2Edge] = {
//        logger.debug(s"SnapshottoEdge: $kv")
    val queryParam = queryRequest.queryParam
    val schemaVer = queryParam.label.schemaVersion
    val snapshotEdgeOpt = snapshotEdgeDeserializer(schemaVer).fromKeyValues(Seq(kv), cacheElementOpt)

    if (isInnerCall) {
      snapshotEdgeOpt.flatMap { snapshotEdge =>
        val edge = snapshotEdge.toEdge.copy(parentEdges = parentEdges)
        if (queryParam.where.map(_.filter(edge)).getOrElse(true)) Option(edge)
        else None
      }
    } else {
      snapshotEdgeOpt.flatMap { snapshotEdge =>
        if (snapshotEdge.allPropsDeleted) None
        else {
          val edge = snapshotEdge.toEdge.copy(parentEdges = parentEdges)
          if (queryParam.where.map(_.filter(edge)).getOrElse(true)) Option(edge)
          else None
        }
      }
    }
  }

  val dummyCursor: Array[Byte] = Array.empty

  def toEdges[K: CanSKeyValue](kvs: Seq[K],
                               queryRequest: QueryRequest,
                               prevScore: Double = 1.0,
                               isInnerCall: Boolean,
                               parentEdges: Seq[EdgeWithScore],
                               startOffset: Int = 0,
                               len: Int = Int.MaxValue): StepResult = {

    val toSKeyValue = implicitly[CanSKeyValue[K]].toSKeyValue _

    if (kvs.isEmpty) StepResult.Empty.copy(cursors = Seq(dummyCursor))
    else {
      val queryOption = queryRequest.query.queryOption
      val queryParam = queryRequest.queryParam
      val labelWeight = queryRequest.labelWeight
      val nextStepOpt = queryRequest.nextStepOpt
      val where = queryParam.where.get
      val label = queryParam.label
      val isDefaultTransformer = queryParam.edgeTransformer.isDefault
      val first = kvs.head
      val kv = first
      val schemaVer = queryParam.label.schemaVersion
      val cacheElementOpt =
        if (queryParam.isSnapshotEdge) None
        else indexEdgeDeserializer(schemaVer).fromKeyValues(Seq(kv), None)

      val (degreeEdges, keyValues) = cacheElementOpt match {
        case None => (Nil, kvs)
        case Some(cacheElement) =>
          val head = cacheElement
          if (!head.isDegree) (Nil, kvs)
          else (Seq(EdgeWithScore(head, 1.0, label)), kvs.tail)
      }

      val lastCursor: Seq[Array[Byte]] = Seq(if (keyValues.nonEmpty) toSKeyValue(keyValues(keyValues.length - 1)).row else dummyCursor)

      if (!queryOption.ignorePrevStepCache) {
        val edgeWithScores = for {
          (kv, idx) <- keyValues.zipWithIndex if idx >= startOffset && idx < startOffset + len
          edge <- (if (queryParam.isSnapshotEdge) toSnapshotEdge(kv, queryRequest, None, isInnerCall, parentEdges) else toEdge(kv, queryRequest, cacheElementOpt, parentEdges)).toSeq
          if where == WhereParser.success || where.filter(edge)
          convertedEdge <- if (isDefaultTransformer) Seq(edge) else convertEdges(queryParam, edge, nextStepOpt)
        } yield {
            val score = edge.rank(queryParam.rank)
            EdgeWithScore(convertedEdge, score, label)
          }
        StepResult(edgeWithScores = edgeWithScores, grouped = Nil, degreeEdges = degreeEdges, cursors = lastCursor)
      } else {
        val degreeScore = 0.0

        val edgeWithScores = for {
          (kv, idx) <- keyValues.zipWithIndex if idx >= startOffset && idx < startOffset + len
          edge <- (if (queryParam.isSnapshotEdge) toSnapshotEdge(kv, queryRequest, None, isInnerCall, parentEdges) else toEdge(kv, queryRequest, cacheElementOpt, parentEdges)).toSeq
          if where == WhereParser.success || where.filter(edge)
          convertedEdge <- if (isDefaultTransformer) Seq(edge) else convertEdges(queryParam, edge, nextStepOpt)
        } yield {
            val edgeScore = edge.rank(queryParam.rank)
            val score = queryParam.scorePropagateOp match {
              case "plus" => edgeScore + prevScore
              case "divide" =>
                if ((prevScore + queryParam.scorePropagateShrinkage) == 0) 0
                else edgeScore / (prevScore + queryParam.scorePropagateShrinkage)
              case _ => edgeScore * prevScore
            }
            val tsVal = processTimeDecay(queryParam, edge)
            val newScore = degreeScore + score
            EdgeWithScore(convertedEdge.copy(parentEdges = parentEdges), score = newScore * labelWeight * tsVal, label = label)
          }

        val sampled =
          if (queryRequest.queryParam.sample >= 0) sample(queryRequest, edgeWithScores, queryRequest.queryParam.sample)
          else edgeWithScores

        val normalized = if (queryParam.shouldNormalize) normalize(sampled) else sampled

        StepResult(edgeWithScores = normalized, grouped = Nil, degreeEdges = degreeEdges, cursors = lastCursor)
      }
    }
  }

  /** End Of Parse Logic */

  protected def toRequestEdge(queryRequest: QueryRequest, parentEdges: Seq[EdgeWithScore]): S2Edge = {
    val srcVertex = queryRequest.vertex
    val queryParam = queryRequest.queryParam
    val tgtVertexIdOpt = queryParam.tgtVertexInnerIdOpt
    val label = queryParam.label
    val labelWithDir = queryParam.labelWithDir
    val (srcColumn, tgtColumn) = label.srcTgtColumn(labelWithDir.dir)
    val propsWithTs = label.EmptyPropsWithTs

    tgtVertexIdOpt match {
      case Some(tgtVertexId) => // _to is given.
        /* we use toSnapshotEdge so dont need to swap src, tgt */
        val src = srcVertex.innerId
        val tgt = tgtVertexId
        val (srcVId, tgtVId) = (SourceVertexId(srcColumn, src), TargetVertexId(tgtColumn, tgt))
        val (srcV, tgtV) = (graph.newVertex(srcVId), graph.newVertex(tgtVId))

        graph.newEdge(srcV, tgtV, label, labelWithDir.dir, propsWithTs = propsWithTs)
      case None =>
        val src = srcVertex.innerId
        val srcVId = SourceVertexId(srcColumn, src)
        val srcV = graph.newVertex(srcVId)

        graph.newEdge(srcV, srcV, label, labelWithDir.dir, propsWithTs = propsWithTs, parentEdges = parentEdges)
    }
  }

  protected def fetchSnapshotEdgeInner(edge: S2Edge): Future[(QueryParam, Option[S2Edge], Option[SKeyValue])] = {
    /* TODO: Fix this. currently fetchSnapshotEdge should not use future cache
      * so use empty cacheKey.
      * */
    val queryParam = QueryParam(labelName = edge.innerLabel.label,
      direction = GraphUtil.fromDirection(edge.labelWithDir.dir),
      tgtVertexIdOpt = Option(edge.tgtVertex.innerIdVal),
      cacheTTLInMillis = -1)
    val q = Query.toQuery(Seq(edge.srcVertex), Seq(queryParam))
    val queryRequest = QueryRequest(q, 0, edge.srcVertex, queryParam)
    //    val q = Query.toQuery(Seq(edge.srcVertex), queryParam)

    fetchSnapshotEdgeKeyValues(queryRequest).map { kvs =>
      val (edgeOpt, kvOpt) =
        if (kvs.isEmpty) (None, None)
        else {
          val snapshotEdgeOpt = toSnapshotEdge(kvs.head, queryRequest, isInnerCall = true, parentEdges = Nil)
          val _kvOpt = kvs.headOption
          (snapshotEdgeOpt, _kvOpt)
        }
      (queryParam, edgeOpt, kvOpt)
    } recoverWith { case ex: Throwable =>
      logger.error(s"fetchQueryParam failed. fallback return.", ex)
      throw new FetchTimeoutException(s"${edge.toLogString}")
    }
  }

  /** end of query */

  /** Mutation Builder */

  /** EdgeMutate */
  def indexedEdgeMutations(edgeMutate: EdgeMutate): Seq[SKeyValue] = {
    // skip sampling for delete operation
    val deleteMutations = edgeMutate.edgesToDeleteWithIndexOpt.flatMap { indexEdge =>
      indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Delete, durability = indexEdge.label.durability))
    }

    val insertMutations = edgeMutate.edgesToInsertWithIndexOpt.flatMap { indexEdge =>
      indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Put, durability = indexEdge.label.durability))
    }

    deleteMutations ++ insertMutations
  }

  def snapshotEdgeMutations(edgeMutate: EdgeMutate): Seq[SKeyValue] =
    edgeMutate.newSnapshotEdge.map(e => snapshotEdgeSerializer(e).toKeyValues.map(_.copy(durability = e.label.durability))).getOrElse(Nil)

  def increments(edgeMutate: EdgeMutate): (Seq[SKeyValue], Seq[SKeyValue]) = {
    (edgeMutate.edgesToDeleteWithIndexOptForDegree.isEmpty, edgeMutate.edgesToInsertWithIndexOptForDegree.isEmpty) match {
      case (true, true) =>
        /* when there is no need to update. shouldUpdate == false */
        Nil -> Nil

      case (true, false) =>
        /* no edges to delete but there is new edges to insert so increase degree by 1 */
        val (buffer, nonBuffer) = EdgeMutate.partitionBufferedIncrement(edgeMutate.edgesToInsertWithIndexOptForDegree)
        buffer.flatMap(buildIncrementsAsync(_)) -> nonBuffer.flatMap(buildIncrementsAsync(_))

      case (false, true) =>
        /* no edges to insert but there is old edges to delete so decrease degree by 1 */
        val (buffer, nonBuffer) = EdgeMutate.partitionBufferedIncrement(edgeMutate.edgesToDeleteWithIndexOptForDegree)
        buffer.flatMap(buildIncrementsAsync(_, -1)) -> nonBuffer.flatMap(buildIncrementsAsync(_, -1))

      case (false, false) =>
        /* update on existing edges so no change on degree */
        Nil -> Nil
    }
  }

  /** IndexEdge */
  def buildIncrementsAsync(indexedEdge: IndexEdge, amount: Long = 1L): Seq[SKeyValue] = {
    val newProps = indexedEdge.updatePropsWithTs()
    newProps.put(LabelMeta.degree.name, new S2Property(indexedEdge.toEdge, LabelMeta.degree, LabelMeta.degree.name, amount, indexedEdge.ts))
    val _indexedEdge = indexedEdge.copy(propsWithTs = newProps)
    indexEdgeSerializer(_indexedEdge).toKeyValues.map(_.copy(operation = SKeyValue.Increment, durability = _indexedEdge.label.durability))
  }

  def buildIncrementsCountAsync(indexedEdge: IndexEdge, amount: Long = 1L): Seq[SKeyValue] = {
    val newProps = indexedEdge.updatePropsWithTs()
    newProps.put(LabelMeta.degree.name, new S2Property(indexedEdge.toEdge, LabelMeta.degree, LabelMeta.degree.name, amount, indexedEdge.ts))
    val _indexedEdge = indexedEdge.copy(propsWithTs = newProps)
    indexEdgeSerializer(_indexedEdge).toKeyValues.map(_.copy(operation = SKeyValue.Increment, durability = _indexedEdge.label.durability))
  }

  //TODO: ServiceColumn do not have durability property yet.
  def buildDeleteBelongsToId(vertex: S2Vertex): Seq[SKeyValue] = {
    val kvs = vertexSerializer(vertex).toKeyValues
    val kv = kvs.head
    vertex.belongLabelIds.map { id =>
      kv.copy(qualifier = Bytes.toBytes(S2Vertex.toPropKey(id)), operation = SKeyValue.Delete)
    }
  }

  def buildVertexPutsAsync(edge: S2Edge): Seq[SKeyValue] = {
    val storeVertex = edge.innerLabel.extraOptions.get("storeVertex").map(_.as[Boolean]).getOrElse(false)

    if (storeVertex) {
      if (edge.op == GraphUtil.operations("delete"))
        buildDeleteBelongsToId(edge.srcForVertex) ++ buildDeleteBelongsToId(edge.tgtForVertex)
      else
        vertexSerializer(edge.srcForVertex).toKeyValues ++ vertexSerializer(edge.tgtForVertex).toKeyValues
    } else {
      Seq.empty
    }
  }

  def buildDegreePuts(edge: S2Edge, degreeVal: Long): Seq[SKeyValue] = {
    edge.propertyInner(LabelMeta.degree.name, degreeVal, edge.ts)
    val kvs = edge.edgesWithIndexValid.flatMap { indexEdge =>
      indexEdgeSerializer(indexEdge).toKeyValues.map(_.copy(operation = SKeyValue.Put, durability = indexEdge.label.durability))
    }

    kvs
  }

  def buildPutsAll(vertex: S2Vertex): Seq[SKeyValue] = {
    vertex.op match {
      case d: Byte if d == GraphUtil.operations("delete") => vertexSerializer(vertex).toKeyValues.map(_.copy(operation = SKeyValue.Delete))
      case _ => vertexSerializer(vertex).toKeyValues.map(_.copy(operation = SKeyValue.Put))
    }
  }

  def info: Map[String, String] = Map("className" -> this.getClass.getSimpleName)

}


text:
// -----------------------------------------------------------------------
// <copyright file="OpportunityContactViewModel.cs" company="Nodine Legal, LLC">
// Licensed to Nodine Legal, LLC under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  Nodine Legal, LLC licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//  http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
// </copyright>
// -----------------------------------------------------------------------

namespace OpenLawOffice.Web.ViewModels.Opportunities
{
    using AutoMapper;
    using OpenLawOffice.Common.Models;

    [MapMe]
    public class OpportunityContactViewModel : CoreViewModel
    {
        public int? Id { get; set; }

        public OpportunityViewModel Opportunity { get; set; }

        public Contacts.ContactViewModel Contact { get; set; }

        public void BuildMappings()
        {
            Mapper.CreateMap<OpenLawOffice.Common.Models.Opportunities.OpportunityContact, OpportunityContactViewModel>()
                .ForMember(dst => dst.IsStub, opt => opt.UseValue(false))
                .ForMember(dst => dst.Created, opt => opt.MapFrom(src => src.Created))
                .ForMember(dst => dst.Modified, opt => opt.MapFrom(src => src.Modified))
                .ForMember(dst => dst.Disabled, opt => opt.MapFrom(src => src.Disabled))
                .ForMember(dst => dst.CreatedBy, opt => opt.ResolveUsing(db =>
                {
                    return new ViewModels.Account.UsersViewModel()
                    {
                        PId = db.CreatedBy.PId,
                        IsStub = true
                    };
                }))
                .ForMember(dst => dst.ModifiedBy, opt => opt.ResolveUsing(db =>
                {
                    return new ViewModels.Account.UsersViewModel()
                    {
                        PId = db.ModifiedBy.PId,
                        IsStub = true
                    };
                }))
                .ForMember(dst => dst.DisabledBy, opt => opt.ResolveUsing(db =>
                {
                    if (db.DisabledBy == null || !db.DisabledBy.PId.HasValue) return null;
                    return new ViewModels.Account.UsersViewModel()
                    {
                        PId = db.DisabledBy.PId.Value,
                        IsStub = true
                    };
                }))
                .ForMember(dst => dst.Id, opt => opt.MapFrom(src => src.Id))
                .ForMember(dst => dst.Opportunity, opt => opt.ResolveUsing(db =>
                {
                    return new ViewModels.Opportunities.OpportunityViewModel()
                    {
                        Id = db.Opportunity.Id,
                        IsStub = true
                    };
                }))
                .ForMember(dst => dst.Contact, opt => opt.ResolveUsing(db =>
                {
                    return new ViewModels.Contacts.ContactViewModel()
                    {
                        Id = db.Contact.Id,
                        IsStub = true
                    };
                }));

            Mapper.CreateMap<OpportunityContactViewModel, OpenLawOffice.Common.Models.Opportunities.OpportunityContact>()
                .ForMember(dst => dst.Created, opt => opt.MapFrom(src => src.Created))
                .ForMember(dst => dst.Modified, opt => opt.MapFrom(src => src.Modified))
                .ForMember(dst => dst.Disabled, opt => opt.MapFrom(src => src.Disabled))
                .ForMember(dst => dst.CreatedBy, opt => opt.ResolveUsing(x =>
                {
                    if (x.CreatedBy == null || !x.CreatedBy.PId.HasValue)
                        return null;
                    return new ViewModels.Account.UsersViewModel()
                    {
                        PId = x.CreatedBy.PId
                    };
                }))
                .ForMember(dst => dst.ModifiedBy, opt => opt.ResolveUsing(x =>
                {
                    if (x.CreatedBy == null || !x.CreatedBy.PId.HasValue)
                        return null;
                    return new ViewModels.Account.UsersViewModel()
                    {
                        PId = x.ModifiedBy.PId
                    };
                }))
                .ForMember(dst => dst.DisabledBy, opt => opt.ResolveUsing(x =>
                {
                    if (x.DisabledBy == null || !x.DisabledBy.PId.HasValue)
                        return null;
                    return new ViewModels.Account.UsersViewModel()
                    {
                        PId = x.DisabledBy.PId.Value
                    };
                }))
                .ForMember(dst => dst.Id, opt => opt.MapFrom(src => src.Id))
                .ForMember(dst => dst.Opportunity, opt => opt.ResolveUsing(model =>
                {
                    if (model.Opportunity == null) return null;
                    return new Common.Models.Opportunities.Opportunity()
                    {
                        Id = model.Opportunity.Id,
                        IsStub = true
                    };
                }))
                .ForMember(dst => dst.Contact, opt => opt.ResolveUsing(model =>
                {
                    if (model.Contact == null) return null;
                    return new Common.Models.Contacts.Contact()
                    {
                        Id = model.Contact.Id,
                        IsStub = true
                    };
                }));
        }
    }
}

redpajama_set_name: RedPajamaGithub


text:
#ifndef	__STATE_MACHINE_H__
#define	__STATE_MACHINE_H__

#include "..\base\State.h"

RDPS_BEGIN
	PDCT_BEGIN

		class Collaborator;

		/********************************************************************************************************************************************
		 *
		 * Classe Resposavel em gerenciar os estados de que a possui.
		 * Faz a troca de estados quando necessario e a execucao deste estado,
		 * tambem da o  comando para tratar as mensagens recebidas.
		 *
		 * Nela consiste:
		 * -> Collaborator *myOwner: uma referencia ao colaborador que possui esse gerenciador;
		 * -> State<Collaborator> *currentState: o estado atual;
		 * -> State<Collaborator> *previousState: o estado anterior;
		 * -> State<Collaborator> *globalState: o estado global.
		 *
		 ********************************************************************************************************************************************/
		class StateMachine
		{
		private:
			Collaborator *myOwner;
			State<Collaborator> *currentState;
			State<Collaborator> *previousState;
			State<Collaborator> *globalState;

		public:
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Construtor
			 * recebe como parametro a referencia de que possui esse gerenciador
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine(Collaborator *owner);
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Construtor de copia
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine(const StateMachine &other);
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Destrutor padrao
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			~StateMachine();

			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que gera uma copia do objeto.
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine *Clone() const;

			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Função de sobrecarga de operador para cópia de dados de outra StateMachine
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &operator=(const StateMachine &other);

			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que retorna a referencia de quem possui esse gerenciador
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			Collaborator *GetOwner() const;
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que retorna o estado atual
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			State<Collaborator> *GetCurrentState()  const;
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que retorna o estado anterior
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			State<Collaborator> *GetPreviousState() const;
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que retorna o estado global
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			State<Collaborator> *GetGlobalState()   const;

			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que seta o estado atual
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &SetCurrentState(State<Collaborator> *_currentState);
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que seta o estado anterior
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &SetPreviousState(State<Collaborator> *_previousState);
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que seta o estado global
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &SetGlobalState(State<Collaborator> *_globalState);

			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que da o comando de execucao dos estados
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &Update();
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que faz a troca de estados.
			 * o estado anterior recebe a referencia do estado atual,
			 * e o atual da o comando exit e troca para o novo estado.
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &ChangeState(State<Collaborator> *newState);
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Reverte para o estado anterior
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			StateMachine &ReverToPreviousState();
			/*---------------------------------------------------------------------------------------------------------------------------------------
			 * Funcao auxiliar que trata as mensagens
			 *---------------------------------------------------------------------------------------------------------------------------------------*/
			bool HandleMessage(const Message &message);
		};

	PDCT_END
RDPS_END

#endif//__STATE_MACHINE_H__


text:
package main

import "net/http"

var authProvider map[string]IAuth

type IAuth interface {
	ServeHTTP(w http.ResponseWriter, r *http.Request)
}

func RegisterAuth(name string, auth IAuth) error {
	return nil
}


text:
'use strict';

describe('where iterator', function () {

	var WhereIterator = require('../../src/iterators/where');
	var ArrayIterator = require('../../src/iterators/array');
	var expect = require('chai').expect;

	it('can filter out elements', function () {

		var where = new WhereIterator(new ArrayIterator([1, 2, 3, 4]), function (value) {
				return value == 2 || value == 4;
			});

		expect(where.realize()).to.eql([2, 4]);
	});

	it('can always get last item at the end', function () {

		var testObject = new WhereIterator(new ArrayIterator([1, 2, 3, 4]), function (value) {
				return value == 2 || value == 4;
			});

		testObject.moveNext();
		testObject.moveNext();
		testObject.moveNext();

		expect(testObject.getCurrent()).to.eql(4);
	});

});


text:
// Copyright lowRISC contributors.
// Licensed under the Apache License, Version 2.0, see LICENSE for details.
// SPDX-License-Identifier: Apache-2.0

#include "sw/device/lib/runtime/print.h"

#include <stdarg.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdint.h>

#include "sw/device/lib/base/macros.h"
#include "sw/device/lib/base/memory.h"
#include "sw/device/lib/base/status.h"

// This is declared as an enum to force the values to be
// compile-time constants, but the type is otherwise not
// used for anything.
enum {
  // Standard format specifiers.
  kPercent = '%',
  kCharacter = 'c',
  kString = 's',
  kSignedDec1 = 'd',
  kSignedDec2 = 'i',
  kUnsignedOct = 'o',
  kUnsignedHexLow = 'x',
  kUnsignedHexHigh = 'X',
  kUnsignedDec = 'u',
  kPointer = 'p',

  // Verilog-style format specifiers.
  kSvBinary = 'b',
  kSvHexLow = 'h',
  kSvHexHigh = 'H',

  // Other non-standard specifiers.
  kHexLeLow = 'y',
  kHexLeHigh = 'Y',
  kStatusResult = 'r',
};

// NOTE: all of the lengths of the strings below are given so that the NUL
// terminator is left off; that way, `sizeof(kConst)` does not include it.
static const char kDigitsLow[16] = "0123456789abcdef";
static const char kDigitsHigh[16] = "0123456789ABCDEF";

static const char kErrorNul[17] = "%<unexpected nul>";
static const char kUnknownSpec[15] = "%<unknown spec>";
static const char kErrorTooWide[12] = "%<bad width>";

static size_t base_dev_null(void *data, const char *buf, size_t len) {
  return len;
}
static buffer_sink_t base_stdout = {
    .data = NULL,
    // Note: Using `&base_dev_null` causes this variable to be placed in the
    // .data section and triggers the assertion in rom.ld.
    .sink = NULL,
};

void base_set_stdout(buffer_sink_t out) {
  if (out.sink == NULL) {
    out.sink = &base_dev_null;
  }
  base_stdout = out;
}

static size_t base_dev_uart(void *data, const char *buf, size_t len) {
  const dif_uart_t *uart = (const dif_uart_t *)data;
  for (size_t i = 0; i < len; ++i) {
    if (dif_uart_byte_send_polled(uart, (uint8_t)buf[i]) != kDifOk) {
      return i;
    }
  }
  return len;
}

void base_uart_stdout(const dif_uart_t *uart) {
  base_set_stdout(
      (buffer_sink_t){.data = (void *)uart, .sink = &base_dev_uart});
}

size_t base_printf(const char *format, ...) {
  va_list args;
  va_start(args, format);
  size_t bytes_left = base_vprintf(format, args);
  va_end(args);
  return bytes_left;
}

size_t base_vprintf(const char *format, va_list args) {
  return base_vfprintf(base_stdout, format, args);
}

typedef struct snprintf_captures_t {
  char *buf;
  size_t bytes_left;
} snprintf_captures_t;

static size_t snprintf_sink(void *data, const char *buf, size_t len) {
  snprintf_captures_t *captures = (snprintf_captures_t *)data;
  if (captures->bytes_left == 0) {
    return 0;
  }

  if (len > captures->bytes_left) {
    len = captures->bytes_left;
  }
  memcpy(captures->buf, buf, len);
  captures->buf += len;
  captures->bytes_left -= len;
  return len;
}

size_t base_snprintf(char *buf, size_t len, const char *format, ...) {
  va_list args;
  va_start(args, format);

  snprintf_captures_t data = {
      .buf = buf,
      .bytes_left = len,
  };
  buffer_sink_t out = {
      .data = &data,
      .sink = &snprintf_sink,
  };
  size_t bytes_left = base_vfprintf(out, format, args);
  va_end(args);
  return bytes_left;
}

size_t base_fprintf(buffer_sink_t out, const char *format, ...) {
  va_list args;
  va_start(args, format);
  size_t bytes_left = base_vfprintf(out, format, args);
  va_end(args);
  return bytes_left;
}

/**
 * Consumes characters from `format` until a '%' or NUL is reached. All
 * characters seen before that are then sinked into `out`.
 *
 * @param out the sink to write bytes to.
 * @param format a pointer to the format string to consume a prefix of.
 * @param[out] bytes_written out param for the number of bytes writen to `out`.
 * @return true if an unprocessed '%' was found.
 */
static bool consume_until_percent(buffer_sink_t out, const char **format,
                                  size_t *bytes_written) {
  size_t text_len = 0;
  while (true) {
    char c = (*format)[text_len];
    if (c == '\0' || c == kPercent) {
      if (text_len > 0) {
        *bytes_written += out.sink(out.data, *format, text_len);
      }
      *format += text_len;
      return c != '\0';
    }
    ++text_len;
  }
}

/**
 * Represents a parsed format specifier.
 */
typedef struct format_specifier {
  char type;
  size_t width;
  char padding;
  bool is_nonstd;
} format_specifier_t;

/**
 * Consumes characters from `format` until a complete format specifier is
 * parsed. See the documentation in `print.h` for full syntax.
 *
 * @param out the sink to write bytes to.
 * @param format a pointer to the format string to consume a prefix of.
 * @param[out] spec out param for the specifier.
 * @return whether the parse succeeded.
 */
static bool consume_format_specifier(buffer_sink_t out, const char **format,
                                     size_t *bytes_written,
                                     format_specifier_t *spec) {
  *spec = (format_specifier_t){0};

  // Consume the percent sign.
  ++(*format);

  // Parse a ! to detect an OpenTitan-specific extension (that isn't one
  // of the Verilog ones...).
  if ((*format)[0] == '!') {
    spec->is_nonstd = true;
    ++(*format);
  }

  // Attempt to parse out an unsigned, decimal number, a "width",
  // after the percent sign; the format specifier is the character
  // immediately after this width.
  //
  // `spec->padding` is used to indicate whether we've seen a width;
  // if nonzero, we have an actual width.
  size_t spec_len = 0;
  while (true) {
    char c = (*format)[spec_len];
    if (c == '\0') {
      *bytes_written += out.sink(out.data, kErrorNul, sizeof(kErrorNul));
      return false;
    }
    if (c < '0' || c > '9') {
      break;
    }
    if (spec->padding == 0) {
      if (c == '0') {
        spec->padding = '0';
        ++spec_len;
        continue;
      }
      spec->padding = ' ';
    }
    spec->width *= 10;
    spec->width += (c - '0');
    ++spec_len;
  }

  if ((spec->width == 0 && spec->padding != 0) || spec->width > 32) {
    *bytes_written += out.sink(out.data, kErrorTooWide, sizeof(kErrorTooWide));
    return false;
  }

  spec->type = (*format)[spec_len];
  *format += spec_len + 1;
  return true;
}

/**
 * Write the digits of `value` onto `out`.
 *
 * @param out the sink to write bytes to.
 * @param value the value to "stringify".
 * @param width the minimum width to print; going below will result in writing
 *        out zeroes.
 * @param base the base to express `value` in.
 * @param glyphs an array of characters to use as the digits of a number, which
 *        should be at least ast long as `base`.
 * @return the number of bytes written.
 */
static size_t write_digits(buffer_sink_t out, uint32_t value, uint32_t width,
                           char padding, uint32_t base, const char *glyphs) {
  // All allocations are done relative to a buffer that could hold the longest
  // textual representation of a number: ~0x0 in base 2, i.e., 32 ones.
  static const int kWordBits = sizeof(uint32_t) * 8;
  char buffer[kWordBits];

  size_t len = 0;
  if (value == 0) {
    buffer[kWordBits - 1] = glyphs[0];
    ++len;
  }
  while (value > 0) {
    uint32_t digit = value % base;
    value /= base;
    buffer[kWordBits - 1 - len] = glyphs[digit];
    ++len;
  }
  width = width == 0 ? 1 : width;
  width = width > kWordBits ? kWordBits : width;
  while (len < width) {
    buffer[kWordBits - len - 1] = padding;
    ++len;
  }
  return out.sink(out.data, buffer + (kWordBits - len), len);
}

static size_t write_status(buffer_sink_t out, status_t value, bool as_json) {
  // The module id is defined to be 3 chars long.
  char mod[] = {'"', 0, 0, 0, '"', ','};
  int32_t arg;
  const char *start;
  bool err = status_extract(value, &start, &arg, &mod[1]);

  // strlen of the status code.
  const char *end = start;
  while (*end)
    end++;
  size_t len = 0;

  len += out.sink(out.data, "{\"", as_json ? 2 : 0);
  len += out.sink(out.data, start, end - start);
  len += out.sink(out.data, "\"", as_json ? 1 : 0);

  len += out.sink(out.data, ":", 1);
  if (err) {
    // All error codes include the module identifier.
    len += out.sink(out.data, "[", 1);
    len += out.sink(out.data, mod, sizeof(mod));
    len += write_digits(out, arg, 0, 0, 10, kDigitsLow);
    len += out.sink(out.data, "]", 1);
  } else {
    // Ok codes include only the arg
    len += write_digits(out, arg, 0, 0, 10, kDigitsLow);
  }
  len += out.sink(out.data, "}", as_json ? 1 : 0);
  return len;
}

/**
 * Hexdumps `bytes` onto `out`.
 *
 * @param out the sink to write bytes to.
 * @param bytes the bytes to dump.
 * @param len the number of bytes to dump.
 * @param width the minimum width to print; going below will result in writing
 *        out zeroes.
 * @param padding the character to use for padding.
 * @param big_endian whether to print in big-endian order (i.e. as %x would).
 * @param glyphs an array of characters to use as the digits of a number, which
 *        should be at least ast long as `base`.
 * @return the number of bytes written.
 */
static size_t hex_dump(buffer_sink_t out, const char *bytes, size_t len,
                       uint32_t width, char padding, bool big_endian,
                       const char *glyphs) {
  size_t bytes_written = 0;
  char buf[32];
  size_t buffered = 0;
  if (len < width) {
    width -= len;
    memset(buf, padding, sizeof(buf));
    while (width > 0) {
      size_t to_write = width > ARRAYSIZE(buf) ? 32 : width;
      bytes_written += out.sink(out.data, buf, to_write);
      width -= to_write;
    }
  }

  for (size_t i = 0; i < len; ++i) {
    size_t idx = big_endian ? len - i - 1 : i;
    buf[buffered] = glyphs[(bytes[idx] >> 4) & 0xf];
    buf[buffered + 1] = glyphs[bytes[idx] & 0xf];
    buffered += 2;

    if (buffered == ARRAYSIZE(buf)) {
      bytes_written += out.sink(out.data, buf, buffered);
      buffered = 0;
    }
  }

  if (buffered != 0) {
    bytes_written += out.sink(out.data, buf, buffered);
  }
  return bytes_written;
}

/**
 * Prints out the next entry in `args` according to `spec`.
 *
 * This function assumes that `spec` accurately describes the next entry in
 * `args`.
 *
 * @param out the sink to write bytes to.
 * @param spec the specifier to use for stringifying.
 * @param[out] bytes_written out param for the number of bytes writen to `out`.
 * @param args the list to pull an entry from.
 */
static void process_specifier(buffer_sink_t out, format_specifier_t spec,
                              size_t *bytes_written, va_list *args) {
  // Switch on the specifier. At this point, we assert that there is
  // an initialized value of correct type in the VA list; if it is
  // missing, the caller has caused UB.
  switch (spec.type) {
    case kPercent: {
      if (spec.is_nonstd) {
        goto bad_spec;
      }
      *bytes_written += out.sink(out.data, "%", 1);
      break;
    }
    case kCharacter: {
      if (spec.is_nonstd) {
        goto bad_spec;
      }
      char value = (char)va_arg(*args, uint32_t);
      *bytes_written += out.sink(out.data, &value, 1);
      break;
    }
    case kString: {
      size_t len = 0;
      if (spec.is_nonstd) {
        // This implements %!s.
        len = va_arg(*args, size_t);
      }

      char *value = va_arg(*args, char *);
      while (!spec.is_nonstd && value[len] != '\0') {
        // This implements %s.
        ++len;
      }

      *bytes_written += out.sink(out.data, value, len);
      break;
    }
    case kSignedDec1:
    case kSignedDec2: {
      if (spec.is_nonstd) {
        goto bad_spec;
      }
      uint32_t value = va_arg(*args, uint32_t);
      if (((int32_t)value) < 0) {
        *bytes_written += out.sink(out.data, "-", 1);
        value = -value;
      }
      *bytes_written +=
          write_digits(out, value, spec.width, spec.padding, 10, kDigitsLow);
      break;
    }
    case kUnsignedOct: {
      if (spec.is_nonstd) {
        goto bad_spec;
      }
      uint32_t value = va_arg(*args, uint32_t);
      *bytes_written +=
          write_digits(out, value, spec.width, spec.padding, 8, kDigitsLow);
      break;
    }
    case kPointer: {
      if (spec.is_nonstd) {
        goto bad_spec;
      }
      // Pointers are formatted as 0x<hex digits>, where the width is always
      // set to the number necessary to represent a pointer on the current
      // platform, that is, the size of uintptr_t in nybbles. For example, on
      // different architecutres the null pointer prints as
      // - rv32imc: 0x00000000 (four bytes, eight nybbles).
      // - amd64:   0x0000000000000000 (eight bytes, sixteen nybbles).
      *bytes_written += out.sink(out.data, "0x", 2);
      uintptr_t value = va_arg(*args, uintptr_t);
      *bytes_written +=
          write_digits(out, value, sizeof(uintptr_t) * 2, '0', 16, kDigitsLow);
      break;
    }
    case kUnsignedHexLow:
      if (spec.is_nonstd) {
        size_t len = va_arg(*args, size_t);
        char *value = va_arg(*args, char *);
        *bytes_written += hex_dump(out, value, len, spec.width, spec.padding,
                                   /*big_endian=*/true, kDigitsLow);
        break;
      }
      OT_FALLTHROUGH_INTENDED;
    case kSvHexLow: {
      uint32_t value = va_arg(*args, uint32_t);
      *bytes_written +=
          write_digits(out, value, spec.width, spec.padding, 16, kDigitsLow);
      break;
    }
    case kUnsignedHexHigh:
      if (spec.is_nonstd) {
        size_t len = va_arg(*args, size_t);
        char *value = va_arg(*args, char *);
        *bytes_written += hex_dump(out, value, len, spec.width, spec.padding,
                                   /*big_endian=*/true, kDigitsHigh);
        break;
      }
      OT_FALLTHROUGH_INTENDED;
    case kSvHexHigh: {
      uint32_t value = va_arg(*args, uint32_t);
      *bytes_written +=
          write_digits(out, value, spec.width, spec.padding, 16, kDigitsHigh);
      break;
    }
    case kHexLeLow: {
      if (!spec.is_nonstd) {
        goto bad_spec;
      }
      size_t len = va_arg(*args, size_t);
      char *value = va_arg(*args, char *);
      *bytes_written += hex_dump(out, value, len, spec.width, spec.padding,
                                 /*big_endian=*/false, kDigitsLow);
      break;
    }
    case kHexLeHigh: {
      if (!spec.is_nonstd) {
        goto bad_spec;
      }
      size_t len = va_arg(*args, size_t);
      char *value = va_arg(*args, char *);
      *bytes_written += hex_dump(out, value, len, spec.width, spec.padding,
                                 /*big_endian=*/false, kDigitsHigh);
      break;
    }
    case kUnsignedDec: {
      if (spec.is_nonstd) {
        goto bad_spec;
      }
      uint32_t value = va_arg(*args, uint32_t);
      *bytes_written +=
          write_digits(out, value, spec.width, spec.padding, 10, kDigitsLow);
      break;
    }
    case kSvBinary: {
      if (spec.is_nonstd) {
        // Bools passed into a ... function will be automatically promoted
        // to int; va_arg(..., bool) is actually UB!
        if (va_arg(*args, int) != 0) {
          *bytes_written += out.sink(out.data, "true", 4);
        } else {
          *bytes_written += out.sink(out.data, "false", 5);
        }
        break;
      }
      uint32_t value = va_arg(*args, uint32_t);
      *bytes_written +=
          write_digits(out, value, spec.width, spec.padding, 2, kDigitsLow);
      break;
    }
    case kStatusResult: {
      status_t value = va_arg(*args, status_t);
      *bytes_written += write_status(out, value, spec.is_nonstd);
      break;
    }
    bad_spec:  // Used with `goto` to bail out early.
    default: {
      *bytes_written += out.sink(out.data, kUnknownSpec, sizeof(kUnknownSpec));
    }
  }
}

size_t base_vfprintf(buffer_sink_t out, const char *format, va_list args) {
  if (out.sink == NULL) {
    out.sink = &base_dev_null;
  }

  // NOTE: This copy is necessary on amd64 and other platforms, where
  // `va_list` is a fixed array type (and, as such, decays to a pointer in
  // an argument list). On PSABI RV32IMC, however, `va_list` is a `void*`, so
  // this is a copy of the pointer, not the array.
  va_list args_copy;
  va_copy(args_copy, args);

  size_t bytes_written = 0;
  while (format[0] != '\0') {
    if (!consume_until_percent(out, &format, &bytes_written)) {
      break;
    }
    format_specifier_t spec;
    if (!consume_format_specifier(out, &format, &bytes_written, &spec)) {
      break;
    }

    process_specifier(out, spec, &bytes_written, &args_copy);
  }

  va_end(args_copy);
  return bytes_written;
}

const char kBaseHexdumpDefaultFmtAlphabet[256] =
    // clang-format off
  // First 32 characters are not printable.
  "................................"
  // Printable ASCII.
  " !\"#$%&'()*+,-./0123456789:;<=>?@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"
  // The rest of the range is also not printable (129 characters).
  "................................................................"
  "................................................................"
  ".";
// clang-format on

static const base_hexdump_fmt_t kBaseHexdumpDefaultFmt = {
    .bytes_per_word = 2,
    .words_per_line = 8,
    .alphabet = &kBaseHexdumpDefaultFmtAlphabet,
};

size_t base_hexdump(const char *buf, size_t len) {
  return base_hexdump_with(kBaseHexdumpDefaultFmt, buf, len);
}

size_t base_snhexdump(char *out, size_t out_len, const char *buf, size_t len) {
  return base_snhexdump_with(out, out_len, kBaseHexdumpDefaultFmt, buf, len);
}

size_t base_fhexdump(buffer_sink_t out, const char *buf, size_t len) {
  return base_fhexdump_with(out, kBaseHexdumpDefaultFmt, buf, len);
}

size_t base_hexdump_with(base_hexdump_fmt_t fmt, const char *buf, size_t len) {
  return base_fhexdump_with(base_stdout, fmt, buf, len);
}

size_t base_snhexdump_with(char *out, size_t out_len, base_hexdump_fmt_t fmt,
                           const char *buf, size_t len) {
  snprintf_captures_t data = {
      .buf = out,
      .bytes_left = out_len,
  };
  buffer_sink_t sink = {
      .data = &data,
      .sink = &snprintf_sink,
  };
  return base_fhexdump_with(sink, fmt, buf, len);
}

size_t base_fhexdump_with(buffer_sink_t out, base_hexdump_fmt_t fmt,
                          const char *buf, size_t len) {
  size_t bytes_written = 0;
  size_t bytes_per_line = fmt.bytes_per_word * fmt.words_per_line;

  for (size_t line = 0; line < len; line += bytes_per_line) {
    bytes_written += base_fprintf(out, "%08x:", line);

    size_t chars_per_line = bytes_per_line * 2 + fmt.words_per_line;
    size_t line_bytes_written = 0;
    for (size_t word = 0; word < bytes_per_line; word += fmt.bytes_per_word) {
      if (len < line + word) {
        char spaces[16] = "                ";
        while (line_bytes_written < chars_per_line) {
          size_t to_print = chars_per_line - line_bytes_written;
          if (to_print > sizeof(spaces)) {
            to_print = sizeof(spaces);
          }
          line_bytes_written += base_fprintf(out, "%!s", to_print, spaces);
        }
        break;
      }

      size_t bytes_left = len - line - word;
      if (bytes_left > fmt.bytes_per_word) {
        bytes_left = fmt.bytes_per_word;
      }
      line_bytes_written += base_fprintf(out, " ");
      line_bytes_written +=
          hex_dump(out, buf + line + word, bytes_left, bytes_left, '0',
                   /*big_endian=*/false, kDigitsLow);
    }
    bytes_written += line_bytes_written;

    bytes_written += base_fprintf(out, "  ");
    size_t buffered = 0;
    char glyph_buffer[16];
    for (size_t byte = 0; byte < bytes_per_line; ++byte) {
      if (buffered == sizeof(glyph_buffer)) {
        bytes_written += base_fprintf(out, "%!s", buffered, glyph_buffer);
        buffered = 0;
      }
      if (line + byte >= len) {
        break;
      }
      glyph_buffer[buffered++] =
          (*fmt.alphabet)[(size_t)(uint8_t)buf[line + byte]];
    }
    if (buffered > 0) {
      bytes_written += base_fprintf(out, "%!s", buffered, glyph_buffer);
    }
    bytes_written += base_fprintf(out, "\n");
  }

  return bytes_written;
}


text:
FOUNDATION_EXPORT double MarvelAPIClientVersionNumber;

//! Project version string for MarvelApiClient.
FOUNDATION_EXPORT const unsigned char MarvelAPIClientVersionString[];

// In this header, you should import all the public headers of your framework using statements like #import <MarvelApiClient/PublicHeader.h>


text:
package com.sachin.risk.common.data.model;

/**
 * @author shicheng.zhang
 * @since 17-7-24 下午4:45
 */
public class PersonalIdData {

    // 前缀6位
    private String prefix;
    // 省份
    private String province;
    // 城市
    private String city;

    public String getPrefix() {
        return prefix;
    }

    public void setPrefix(String prefix) {
        this.prefix = prefix;
    }

    public String getProvince() {
        return province;
    }

    public void setProvince(String province) {
        this.province = province;
    }

    public String getCity() {
        return city;
    }

    public void setCity(String city) {
        this.city = city;
    }

    @Override
    public String toString() {
        return "PersonalIdData{" +
                "prefix='" + prefix + '\'' +
                ", province='" + province + '\'' +
                ", city='" + city + '\'' +
                '}';
    }
}


text:
<!DOCTYPE html>

    <meta charset="utf-8" />
    <title>statsmodels.genmod.qif.QIFResults — statsmodels v0.10.0 documentation</title>
    <link rel="stylesheet" href="../_static/nature.css" type="text/css" />
    <link rel="stylesheet" href="../_static/pygments.css" type="text/css" />
    <link rel="stylesheet" type="text/css" href="../_static/graphviz.css" />

    <link rel="shortcut icon" href="../_static/statsmodels_hybi_favico.ico"/>
    <link rel="author" title="About these documents" href="../about.html" />
    <link rel="index" title="Index" href="../genindex.html" />
    <link rel="search" title="Search" href="../search.html" />
    <link rel="next" title="statsmodels.genmod.qif.QIFResults.aic" href="statsmodels.genmod.qif.QIFResults.aic.html" />
    <link rel="prev" title="statsmodels.genmod.generalized_estimating_equations.GEEMargins.summary_frame" href="statsmodels.genmod.generalized_estimating_equations.GEEMargins.summary_frame.html" />
<link rel="stylesheet" href="../_static/examples.css" type="text/css" />
<link rel="stylesheet" href="../_static/facebox.css" type="text/css" />

Navigation

          index

          modules |

          next |

          previous |

Install
  |

Support
  |

Bugs
  |

Develop
  |

Examples
  |

FAQ
  |

Generalized Estimating Equations |

statsmodels.genmod.qif.QIFResults¶

class statsmodels.genmod.qif.QIFResults(model, params, cov_params, scale, use_t=False, **kwds)[source]¶

Results class for QIF Regression

Methods

<colgroup>
<col style="width: 10%" />
<col style="width: 90%" />
</colgroup>

aic()

An AIC-like statistic for models fit using QIF.

bic()

A BIC-like statistic for models fit using QIF.

bse()

The standard errors of the parameter estimates.

conf_int([alpha, cols, method])

Returns the confidence interval of the fitted parameters.

cov_params([r_matrix, column, scale, cov_p, …])

Returns the variance/covariance matrix.

f_test(r_matrix[, cov_p, scale, invcov])

Compute the F-test for a joint linear hypothesis.

fittedvalues()

Returns the fitted values from the model.

initialize(model, params, **kwd)

Initialize (possibly re-initialize) a Results instance.

llf()

Log-likelihood of model

load(fname)

load a pickle, (class method)

normalized_cov_params()

See specific model class docstring

predict([exog, transform])

Call self.model.predict with self.params as the first argument.

pvalues()

The two-tailed p values for the t-stats of the params.

remove_data()

remove data arrays, all nobs arrays from result and model

save(fname[, remove_data])

save a pickle of this instance

summary([yname, xname, title, alpha])

Summarize the QIF regression results

t_test(r_matrix[, cov_p, scale, use_t])

Compute a t-test for a each linear hypothesis of the form Rb = q

t_test_pairwise(term_name[, method, alpha, …])

perform pairwise t_test with multiple testing corrected p-values

tvalues()

Return the t-statistic for a given parameter estimate.

wald_test(r_matrix[, cov_p, scale, invcov, …])

Compute a Wald-test for a joint linear hypothesis.

wald_test_terms([skip_single, …])

Compute a sequence of Wald tests for terms over multiple columns

Previous topic

statsmodels.genmod.generalized_estimating_equations.GEEMargins.summary_frame

Next topic

statsmodels.genmod.qif.QIFResults.aic

This Page

Show Source

Quick search

    <form class="search" action="../search.html" method="get">
      <input type="text" name="q" aria-labelledby="searchlabel" />
      <input type="submit" value="Go" />
    </form>

        © Copyright 2009-2018, Josef Perktold, Skipper Seabold, Jonathan Taylor, statsmodels-developers.
      Created using Sphinx 2.1.2.


text:
require 'spec_helper'

describe PDC do
  before do
    VCR.insert_cassette fixture_name
  end

  after do
    VCR.eject_cassette
  end

  let(:arch) { PDC::V1::Arch }

  it 'has an attribute name' do
    arch.attributes.must_equal ['name']
  end

  it 'returns the total count and not items in page' do
    total_count = arch.count
    page_count = arch.page(2).count
    page_count.must_equal total_count

    arch_in_page = arch.page(2).contents!
    arch_in_page.length.must_be :<, page_count
  end

  it '.page.count returns total count' do
    pg2 = arch.page(2)
    pg2_count = pg2.all.length

    arches_count = pg2.count # must return all arches count
    arches_count.wont_equal pg2_count
  end
end


text: This folder contains the Main model, view and controller.


text:
<?php
use yii\helpers\Url;

echo '
Distribution of Interaction
';
echo '
';
      if($interactions_by_type_json_table && ($total_interactions > 0)){
          $this->registerJs("GoogleCharts.drawCircle(".$interactions_by_type_json_table.", 'tw', 'interactions_by_type')", yii\web\View::POS_END);
		  echo '

';
	}else{
    	echo '

';
    }
echo '
';


text:
[Flask](http://flask.pocoo.org) is a lightweight framework for Python web apps.

To install it, just run `pip install flask` in your Terminal.

### Step 2: create basic folders for your new app

Create a new directory with your app's name with two subdirectories:

		app/
			templates/  # will be used for content files
			static/		# will be used for images and other stuff that doesn't really change that often

### Step 3: create a basic web app with Flask

Create a file called `server.py` in you main project folder (`app/`):

---

	from flask import Flask, render_template

	app = Flask(__name__)

	@app.route('/')
	def index():
		return "Hi there!"

	if __name__ == '__main__':
		app.run(debug=True)

---

From Terminal, run `server.py` with python (`python server.py`). Make sure you're in the right directory.

Open `http://localhost:5000` in your browser to see your web app.

You can add other app routes. For example:

	@app.route('/test')
	def test():
		return "This is just a test"

### Step 4: create a basic template for your web app

Create a basic HTML file. Call it `base.html`, save it in your `templates/ folder`.

	<!DOCTYPE html>

			<title>My fancy new app</title>

This is a big heading

			{% block content %}
			{% endblock %}

### Step 5: create some other templates

Now that you have your base, you can create html files with content for each page of your web app. You'll need a starting page first. Create a file named `index.html` in your `templates` folder:

	{% extends "base.html" %}
	{% block content %}

Hello world.

	{% endblock %}

Now tell your `server.py` page to render this template:

	@app.route('/')
	def index():
		return render_template('index.html')

Rinse and repeat for every page you want to add. Create more files in your `templates` folder with names like: `page_name.html`, following the structure of `index.html`. When you're done, don't forget to tell `server.py` how you want your app to route. Example:

	@app.route('/page_name')
	def page_name():
		return render_template('page_name.html')

Hint: your new page name appears three times in the code above. Don't forget to change all references when adding a new page.

### Step 6: add Boostrap for a better look

Want your web app to look better without much effort? Download [Bootstrap](http://getbootstrap.com). Unzip. Rename directory to `bootstrap`, move it to folder `app/static`.

Edit your `base.html` again; you have to tell our basic template where your CSS file is located and we're also adding the meta viewport tag to make the web app behave well on mobile devices:

	<!DOCTYPE html>

			<meta name="viewport" content="width=device-width, initial-scale=1.0">

			<link href="{{ url_for('.static', filename='bootstrap/css/bootstrap.css') }}" rel="stylesheet" />

			<title>Beautiful web</title>

### Step 7: explore Bootstrap

[Bootstrap CSS documentation](http://getbootstrap.com/css/) is your new best friend.

Some cool things you can do:

- Use the [grid system](http://getbootstrap.com/css/#grid) to create rows with div boxes taking up to 12 columns per row. For instance, you can have a row with just one large box:

	`

Some content

`

- A row divided in half with two boxes:

	`

First half

Second half

`

- … and [much more](http://examples.getbootstrap.com/grid/).

- Add a cool looking [navigation bar](http://getbootstrap.com/components/#navbar) on top of every page.

### Step 8: add JavaScript from Bootstrap

To make the page even more awesome and dynamic, you can also use Bootstrap's JavaScript. Add the following code right before your ending `</body>` tag in `base.html`:

And you'll also need to [download jquery](http://jquery.com/download/). Create a `js` folder in `app/static` and move the downloaded `jquery-1.10.2.min.js` in there. In `base.html`, add this line before the Bootstrap JavaScript tag:

The final lines of your `base.html` should now look like this:

### Step 9: play with Bootstrap's jquery plugins

Take a look at available [jquery plugins](http://getbootstrap.com/javascript/). Choose one, copy the example code and have fun with it!

### Step 10: custom CSS

Want to change the size of paragraphs, the color of a class or something else? Add your own custom CSS to override Bootstrap's defaults. Create a `css` folder in `app/static` and a new, blank `custom.css` file. Add any custom CSS definition:

	p {
		color:#FF40FF;
	}

Edit `base.html` and add the following link to your custom CSS file after the link to Bootstrap's CSS in the `<head>` part of the file:

	<link href="{{ url_for('.static', filename='css/custom.css') }}" rel="stylesheet" />

### What next?

Make it pretty, play with it, have fun!


text:
using System.IO;
using Mono.Cecil;

namespace ILspect.Model
{
    public class MemberDisassembly
    {
        public string Name => Definition.Name;
        public MemberType MemberType { get; }
        public IMemberDefinition Definition { get; }

        public MemberDisassembly(IMemberDefinition memberDefinition)
        {
            Definition = memberDefinition;
            MemberType = GetMemberType(memberDefinition);
        }

        private static MemberType GetMemberType(IMemberDefinition memberDefinition)
        {
            switch (memberDefinition)
            {
                case TypeDefinition _: return MemberType.NestedType;
                case FieldDefinition _: return MemberType.Field;
                case PropertyDefinition _: return MemberType.Property;
                case EventDefinition _: return MemberType.Event;
                case MethodDefinition _: return MemberType.Method;
                default:
                    throw new InvalidDataException($"Unknown member type: {memberDefinition.GetType().FullName}");
            }
        }
    }
}


text:
#!/usr/bin/env kmk_ash
#
# CDDL HEADER START
#
# The contents of this file are subject to the terms of the
# Common Development and Distribution License, Version 1.0 only
# (the "License").  You may not use this file except in compliance
# with the License.
#
# You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
# or http://www.opensolaris.org/os/licensing.
# See the License for the specific language governing permissions
# and limitations under the License.
#
# When distributing Covered Code, include this CDDL HEADER in each
# file and include the License file at usr/src/OPENSOLARIS.LICENSE.
# If applicable, add the following below this CDDL HEADER, with the
# fields enclosed by brackets "[]" replaced with your own identifying
# information: Portions Copyright [yyyy] [name of copyright owner]
#
# CDDL HEADER END
#
#
# Copyright 2003 Sun Microsystems, Inc.  All rights reserved.
# Use is subject to license terms.
#
#ident	"%Z%%M%	%I%	%E% SMI"

echo "\
\n\
\n\
#pragma ident\t\"%Z%%M%\t%I%\t%E% SMI\"\n\
\n\
#include <dt_errtags.h>
\n\
static const char *const _dt_errtags[] = {"

# WARNING: Tabs in expressions.
#pattern='^	\(D_[A-Z0-9_]*\),*'
#replace='	"\1",'
#
#sed -n "s/$pattern/$replace/p" || exit 1
set -x
kmk_sed --lang_c -n 's/^[[:space:]]*\(D_[A-Z0-9_]*\),*/	"\1",/p' $1 || exit 1

echo "\
};\n\
\n\
static const int _dt_ntag = sizeof (_dt_errtags) / sizeof (_dt_errtags[0]);\n\
\n\
const char *
dt_errtag(dt_errtag_t tag)
{
	return (_dt_errtags[(tag > 0 && tag < _dt_ntag) ? tag : 0]);
}"

exit 0


text:
import { Component } from '@angular/core';

@Component({
  selector: 'home',
  template: `

Home

  `
})
export class HomeComponent {}


text:
class NtpBackgroundService;
class Profile;

class NtpBackgroundServiceFactory : public ProfileKeyedServiceFactory {
 public:
  // Returns the NtpBackgroundService for |profile|.
  static NtpBackgroundService* GetForProfile(Profile* profile);

  static NtpBackgroundServiceFactory* GetInstance();

  NtpBackgroundServiceFactory(const NtpBackgroundServiceFactory&) = delete;
  NtpBackgroundServiceFactory& operator=(const NtpBackgroundServiceFactory&) =
      delete;

 private:
  friend struct base::DefaultSingletonTraits<NtpBackgroundServiceFactory>;

  NtpBackgroundServiceFactory();
  ~NtpBackgroundServiceFactory() override;

  // Overridden from BrowserContextKeyedServiceFactory:
  KeyedService* BuildServiceInstanceFor(
      content::BrowserContext* profile) const override;
};

#endif  // CHROME_BROWSER_SEARCH_BACKGROUND_NTP_BACKGROUND_SERVICE_FACTORY_H_